Go 并发编程：通道常见应用范式

通道经典应用

一、闭包实现通道访问限制

在Go的并发编程中，创建通道和开辟协程是非常方便且容易的，正因如此，有可能会导致开发者滥用。如果在团队开发中没有良好的协商和规范，更可能会导致并发数据不安全。例如：

func Demo() {
	ch := make(chan int, 0)
	go dosomething(ch, 10)
	go dosomething(ch, 20)
	dosomething(ch, 30)
}

func dosomething(ch chan int, num int) {
	for i := 1; i < num; i++ {
		ch <- i
	}
	close(ch1)
}

以上是个非常明显的不规范使用通道的例子，你能得到的只有死锁！

fatal error: all goroutines are asleep - deadlock!

那么能不能限制通道的使用呢，即特定通道只让特定协程使用？有几种解决方案：

团队协商，在代码规范上下功夫，显然这是不安全的。
使用同步功能
闭包实现访问受限的通道

以下我们使用第三种方案：闭包实现访问受限的通道，只允许特定协程使用

func Demo() {
	// 生产者：producter 内部开辟一条协程往里面发送数据，并返回一个只读通道
	producter := func() <-chan int {
		results := make(chan int, 5) // 该通道只作用于特定闭包内的作用域
		go func() {
			defer close(results)
			for i := 0; i <= 5; i++ {
				results <- i
			}
		}()
		return results
	}

	// 消费者：运行这个闭包时需要传入一个只读通道
	consumer := func(results <-chan int) { 
		for result := range results {
			fmt.Printf("Received: %d\n", result)
		}
		fmt.Println("Done receiving!")
	}
        
 	consumer(producter())
}

二、for-select范式：关于多通道操作的整合

我们知道select语句是go专门为多通道操作提供的原语，单个select语句可以一次性的从多个通道选取一个来读写，只要哪个通道先不处于阻塞状态便选取哪个通道读写。而结合for循环语句构成的for-select结构可以循环不断的从多通道读写数据，直到特定条件退出。

for-select循环模式如下所示：

for { // 无限循环或遍历
    select {
    // 对通道进行操作
    }
}

常见的几种for-select循环的用法: a. 在通道上发送迭代变量

for _, s := range []string{"a", "b", "c"} {
    select {
    case <-done:
        return
    case stringStream <- s:   // slice数据循环迭代写入channel
    }
}

b. 无限循环等待停止

// 第一种方式
for {
    select {
    case <-done: 
        return   // 停止返回
    default:
    }
    // 执行非抢占任务
}

// 第二种方式
for {
    select {
    case <-done:
        return 
    default:    
      // 将要执行的任务放入default分支中
      // 执行非抢占任务
    }
}

通过善用通道，我们可以在许多并发过程中尽量避免使用同步锁，select原语可以集中处理多个通道，大大提高了开发和运行效率。

三、or-channel ：递归多个通道的或读取，只要有一个通道返回即完成


/*
递归多个通道的或读取，只要有一个通道返回即完成
*/
func Demo() {
	var or func(channels ...<-chan interface{}) <-chan interface{}
    // 建立了名为or的递归函数，接收数量可变的通道并返回单个通道。
	or = func(channels ...<-chan interface{}) <-chan interface{} 
    // 两个递归终止条件		
    switch len(channels) {
		case 0:  // 如果传入的切片是空的，我们简单的返回一个nil通道
			return nil
		case 1:  // 如果切片只含有一个元素，我们就返回给元素
			return channels[0]
		}
      
		orChannel := make(chan interface{})
        // 建立一个goroutine，以便可以不受阻塞地等待我们通道上的消息
		go func() { 
			defer close(orDone)

			switch len(channels) {
			case 2: // 由于我们这里是递归的，每次递归调用将至少有两个通道。作为保持goroutine数量受到限制的优化方法，们在这里为仅使用两个通道的时设置了一个特殊情况。
				select {
				case <-channels[0]:
				case <-channels[1]:
				}
			default: // 递归地在第三个索引之后，从切片中的所有通道中创建一个or通道，然后从中选择。递归操作会逐层累计直到取到第一个通道元素。我们在其中传递了orChannel通道，这样当该树状结构顶层的goroutines退出时，结构底层的goroutines也会退出。
				select {
				case <-channels[0]:
				case <-channels[1]:
				case <-channels[2]:
				case <-or(append(channels[3:], orChannel)...): 
				}
			}
		}()
		return orDone
	}



	// 下面这个例子将经过一段时间后关闭通道，然后使用or函数将这些通道合并到一个关闭的通道中：
	sig := func(after time.Duration) <-chan interface{} { // 创建了一个通道，当后续时间中指定的时间结束时将关闭该通道
		c := make(chan interface{})
		go func() {
			defer close(c)
			time.Sleep(after)
		}()
		return c
	}

	start := time.Now() // 设置追踪自or函数的通道开始阻塞的起始时间
	<-or(sig(2*time.Hour), sig(5*time.Minute), sig(1*time.Second), sig(1*time.Hour), sig(1*time.Minute))
	fmt.Printf("done after %v", time.Since(start)) // 打印阻塞发生的时间
}

这是一种奇妙的做法，你可以将任意数量的通道组合到单个通道中，只要任何作为组件的通道关闭或被写入，整个通道就会关闭。

四、chRange 封装安全的通道遍历读取

有时你会与来自系统不同部分的通道交互。与管道不同的是，当你使用的代码通过done通道取消操作时，你无法对通道的行为方式做出判断。也就是说，你不知道正在执行读取操作的goroutine现在是什么状态。出于这个原因，正如我们在“防止Goroutine泄漏”中所阐述的那样，需要用select语句来封装我们的读取操作和done通道。可以简单的写成这样：

for val := range myChan {
	// 对 val 进行处理
}

展开后可以写成这样：

loop:
	for {
		select {
		case <-done:
			break loop
		case maybeVal, ok := <-myChan:
			if ok == false {
				return // or maybe break from for
			}
			// Do something with val
		}
	}

这样做可以快速退出嵌套循环。继续使用goroutines编写更清晰的并发代码，而不是过早优化的主题，我们可以用一个goroutine来解决这个问题。我们封装了细节，以便其他人调用更方便：

/*
封装一个通用的安全的通道读取器，以便于可安全地for range遍历任意通道
*/
var chRange = func(done, ch <-chan interface{}) <-chan interface{} {
	valStream := make(chan interface{})
    // 使用协程闭包封装安全的读取通道 
	go func() {
		defer close(valStream)
		for {
			select {
			case <-done:
				return
			case v, ok := <-ch:
				if ok == false {
					return
				}
				select {
				case valStream <- v:
				case <-done:
				}
			}
		}
	}()

	return valStream
}

调用示例：这样对任意通道我们都可以简单安全的读取

func Demo() {
	done := make(chan interface{})
	defer close(done)

	ch := make(chan interface{})
	go func() {
		defer close(ch)
		for i := 0; i < 10; i++ {
			ch <- i
		}
	}()

	for val := range chRange(done, ch) {
		fmt.Printf("read %v \n", val)
	}
}

五、tee-channel 分割通道数据流

tee-channel类似Linux的tee命令，分割来自通道的值，以便将它们发送到两个独立区域。想象一下：你可能想要在一个通道上接收一系列操作指令，将它们发送给执行者，同时记录操作日志。

var tee = func(done <-chan interface{}, in <-chan interface{}) (_, _ <-chan interface{}) {

	out1 := make(chan interface{})
	out2 := make(chan interface{})

	go func() {
		defer close(out1)
		defer close(out2)
		for val := range chRange(done, in) {
			select {
			case <-done:
			default:
				out1 <- val
				out2 <- val
			}

		}
	}()
	return out1, out2
}

注意写入out1和out2是紧密耦合的。直到out1和out2都被写入，迭代才能继续。通常这不是问题，因为无论如何，处理来自每个通道的读取流程的吞吐量应该是tee之外的关注点，但值得注意。这是一个快速调用示例：

func Demo()  {
	done := make(chan interface{})
	defer close(done)

	ch := make(chan interface{})
	go func() {
		for i := 0; i < 10; i++ {
			ch <- i
		}
		defer close(ch)
	}()

	out1, out2 := tee(done, ch)

	for val1 := range out1 {
		fmt.Printf("out1: %v, out2: %v\n", val1, <-out2)
	}

利用这种模式，很容易使用通道作为系统数据的连接点。

六、bridge-channel

在某些情况下，你可能会发现自己想要使用一系列通道,即你可能需要从一个通道中获取多个通道的值：

<-chan <-chan interface{}

这与将某个通道的数据切片合并到一个通道中稍有不同，这种调用方式意味着一系列通道有序的写入操作。从通道读取一系列通道的值，类似多通道过独木桥。

// 通道桥接
var bridge = func(done <-chan interface{}, chanStream <-chan <-chan interface{}) <-chan interface{} {

	valStream := make(chan interface{}) // 1
	go func() {
		defer close(valStream)
		for { // 2
			var stream <-chan interface{}
			select {
			case maybeStream, ok := <-chanStream:
				if ok == false {
					return
				}
				stream = maybeStream
			case <-done:
				return
			}
			for val := range chDone(done, stream) { // 3
				select {
				case valStream <- val:
				case <-done:
				}
			}
		}
	}()
	return valStream
}

使用示例：

func Demo() {
	genVals := func() <-chan <-chan interface{} {

		chanStream := make(chan (<-chan interface{}))

		go func() {
			defer close(chanStream)
			for i := 0; i < 10; i++ {
				stream := make(chan interface{}, 1)
				stream <- i
				close(stream)
				chanStream <- stream
			}
		}()
		return chanStream
	}

	for v := range bridge(nil, genVals()) {
		fmt.Printf("%v ", v)
	}
}