《快学 Go 语言》第 12 课 —— 通道

不一样的并行协程之间交流的方式有两种，一种是经过共享变量，另外一种是经过队列。Go 语言鼓励使用队列的形式来交流，它单独为协程之间的队列数据交流定制了特殊的语法 —— 通道。

通道是协程的输入和输出。做为协程的输出，通道是一个容器，它能够容纳数据。做为协程的输入，通道是一个生产者，它能够向协程提供数据。通道做为容器是有限定大小的，满了就写不进去，空了就读不出来。通道还有它本身的类型，它能够限定进入通道的数据的类型。

建立通道

建立通道只有一种语法，那就是 make 全局函数，提供第一个类型参数限定通道能够容纳的数据类型，再提供第二个整数参数做为通道的容器大小。大小参数是可选的，若是不填，那这个通道的容量为零，叫着「非缓冲型通道」，非缓冲型通道必须确保有协程正在尝试读取当前通道，不然写操做就会阻塞直到有其它协程来从通道中读东西。非缓冲型通道老是处于既满又空的状态。与之对应的有限定大小的通道就是缓冲型通道。在 Go 语言里不存在无界通道，每一个通道都是有限定最大容量的。

// 缓冲型通道，里面只能放整数
var bufferedChannel = make(chan int, 1024)
// 非缓冲型通道
var unbufferedChannel = make(chan int)
复制代码

读写通道

Go 语言为通道的读写设计了特殊的箭头语法糖 <-，让咱们使用通道时很是方便。把箭头写在通道变量的右边就是写通道，把箭头写在通道的左边就是读通道。一次只能读写一个元素。

package main

import "fmt"

func main() {
	var ch chan int = make(chan int, 4)
	for i:=0; i<cap(ch); i++ {
		ch <- i   // 写通道
	}
	for len(ch) > 0 {
		var value int = <- ch  // 读通道
		fmt.Println(value)
	}
}
复制代码

通道做为容器，它能够像切片同样，使用 cap() 和 len() 全局函数得到通道的容量和当前内部的元素个数。通道通常做为不一样的协程交流的媒介，在同一个协程里它也是可使用的。

读写阻塞

通道满了，写操做就会阻塞，协程就会进入休眠，直到有其它协程读通道挪出了空间，协程才会被唤醒。若是有多个协程的写操做都阻塞了，一个读操做只会唤醒一个协程。

通道空了，读操做就会阻塞，协程也会进入睡眠，直到有其它协程写通道装进了数据才会被唤醒。若是有多个协程的读操做阻塞了，一个写操做也只会唤醒一个协程。

package main

import "fmt"
import "time"
import "math/rand"

func send(ch chan int) {
	for {
		var value = rand.Intn(100)
		ch <- value
		fmt.Printf("send %d\n", value)
	}
}

func recv(ch chan int) {
	for {
		value := <- ch
		fmt.Printf("recv %d\n", value)
		time.Sleep(time.Second)
	}
}

func main() {
	var ch = make(chan int, 1)
	// 子协程循环读
	go recv(ch)
	// 主协程循环写
	send(ch)
}

--------
send 81
send 87
recv 81
recv 87
send 47
recv 47
send 59
复制代码

关闭通道

Go 语言的通道有点像文件，不但支持读写操做， 还支持关闭。读取一个已经关闭的通道会当即返回通道类型的「零值」，而写一个已经关闭的通道会抛异常。若是通道里的元素是整型的，读操做是不能经过返回值来肯定通道是否关闭的。

package main

import "fmt"

func main() {
	var ch = make(chan int, 4)
	ch <- 1
	ch <- 2
	close(ch)

	value := <- ch
	fmt.Println(value)
	value = <- ch
	fmt.Println(value)
	value = <- ch
	fmt.Println(value)
}

-------
1
2
0
复制代码

这时候就须要引入一个新的知识点 —— 使用 for range 语法糖来遍历通道

for range 语法咱们已经见了不少次了，它是多功能的，除了能够遍历数组、切片、字典，还能够遍历通道，取代箭头操做符。当通道空了，循环会暂停阻塞，当通道关闭时，阻塞中止，循环也跟着结束了。当循环结束时，咱们就知道通道已经关闭了。

package main

import "fmt"

func main() {
	var ch = make(chan int, 4)
	ch <- 1
	ch <- 2
	close(ch)

 // for range 遍历通道
	for value := range ch {
		fmt.Println(value)
	}
}

------
1
2
复制代码

通道若是没有显式关闭，当它再也不被程序使用的时候，会自动关闭被垃圾回收掉。不过优雅的程序应该将通道当作资源，显式关闭每一个再也不使用的资源是一种良好的习惯。

通道写安全

上面提到向一个已经关闭的通道执行写操做会抛出异常，这意味着咱们在写通道时必定要确保通道没有被关闭。

package main

import "fmt"

func send(ch chan int) {
	i := 0
	for {
		i++
		ch <- i
	}
}

func recv(ch chan int) {
	value := <- ch
	fmt.Println(value)
	value = <- ch
	fmt.Println(value)
	close(ch)
}

func main() {
	var ch = make(chan int, 4)
	go recv(ch)
	send(ch)
}

---------
1
2
panic: send on closed channel

goroutine 1 [running]:
main.send(0xc42008a000)
	/Users/qianwp/go/src/github.com/pyloque/practice/main.go:9 +0x44
main.main()
	/Users/qianwp/go/src/github.com/pyloque/practice/main.go:24 +0x66
exit status 2
复制代码

那如何确保呢？Go 语言并不存在一个内置函数能够判断出通道是否已经被关闭。即便存在这样一个函数，当你判断时通道没有关闭，并不意味着当你往通道里写数据时它就必定没有被关闭，并发环境下，它是可能被其它协程随时关闭的。

确保通道写安全的最好方式是由负责写通道的协程本身来关闭通道，读通道的协程不要去关闭通道。

package main

import "fmt"

func send(ch chan int) {
 ch <- 1
 ch <- 2
 ch <- 3
 ch <- 4
 close(ch)
}

func recv(ch chan int) {
 for v := range ch {
  fmt.Println(v)
 }
}

func main() {
 var ch = make(chan int, 1)
 go send(ch)
 recv(ch)
}

-----------
1
2
3
4
复制代码

这个方法确实能够解决单写多读的场景，可要是赶上了多写单读的场合该怎么办呢？任意一个读写通道的协程都不能够随意关闭通道，不然会致使其它写通道协程抛出异常。这时候就必须让其它不相干的协程来干这件事，这个协程须要等待全部的写通道协程都结束运行后才能关闭通道。那其它协程要如何才能知道全部的写通道已经结束运行了呢？这个就须要使用到内置 sync 包提供的 WaitGroup 对象，它使用计数来等待指定事件完成。

package main

import "fmt"
import "time"
import "sync"

func send(ch chan int, wg *sync.WaitGroup) {
	defer wg.Done() // 计数值减一
	i := 0
	for i < 4 {
		i++
		ch <- i
	}
}

func recv(ch chan int) {
	for v := range ch {
		fmt.Println(v)
	}
}

func main() {
	var ch = make(chan int, 4)
	var wg = new(sync.WaitGroup)
	wg.Add(2) // 增长计数值
	go send(ch, wg)  // 写
	go send(ch, wg)  // 写
	go recv(ch)
	// Wait() 阻塞等待全部的写通道协程结束
 // 待计数值变成零，Wait() 才会返回
 wg.Wait()
	// 关闭通道
 close(ch)
 time.Sleep(time.Second)
}

---------
1
2
3
4
1
2
3
4
复制代码

多路通道

在真实的世界中，还有一种消息传递场景，那就是消费者有多个消费来源，只要有一个来源生产了数据，消费者就能够读这个数据进行消费。这时候能够将多个来源通道的数据汇聚到目标通道，而后统一在目标通道进行消费。

package main

import "fmt"
import "time"

// 每隔一会生产一个数
func send(ch chan int, gap time.Duration) {
	i := 0
	for {
		i++
		ch <- i
		time.Sleep(gap)
	}
}

// 将多个原通道内容拷贝到单一的目标通道
func collect(source chan int, target chan int) {
	for v := range source {
		target <- v
	}
}

// 从目标通道消费数据
func recv(ch chan int) {
	for v := range ch {
		fmt.Printf("receive %d\n", v)
	}
}


func main() {
	var ch1 = make(chan int)
	var ch2 = make(chan int)
	var ch3 = make(chan int)
	go send(ch1, time.Second)
	go send(ch2, 2 * time.Second)
	go collect(ch1, ch3)
	go collect(ch2, ch3)
	recv(ch3)
}

---------
receive 1
receive 1
receive 2
receive 2
receive 3
receive 4
receive 3
receive 5
receive 6
receive 4
receive 7
receive 8
receive 5
receive 9
....

复制代码

可是上面这种形式比较繁琐，须要为每一种消费来源都单独启动一个汇聚协程。Go 语言为这种使用场景带来了「多路复用」语法糖，也就是下面要讲的 select 语句，它能够同时管理多个通道读写，若是全部通道都不能读写，它就总体阻塞，只要有一个通道能够读写，它就会继续。下面咱们使用 select 语句来简化上面的逻辑

package main

import "fmt"
import "time"

func send(ch chan int, gap time.Duration) {
	i := 0
	for {
		i++
		ch <- i
		time.Sleep(gap)
	}
}

func recv(ch1 chan int, ch2 chan int) {
	for {
		select {
			case v := <- ch1:
				fmt.Printf("recv %d from ch1\n", v)
			case v := <- ch2:
				fmt.Printf("recv %d from ch2\n", v)
		}
	}
}

func main() {
	var ch1 = make(chan int)
	var ch2 = make(chan int)
	go send(ch1, time.Second)
	go send(ch2, 2 * time.Second)
	recv(ch1, ch2)
}

------------
recv 1 from ch2
recv 1 from ch1
recv 2 from ch1
recv 3 from ch1
recv 2 from ch2
recv 4 from ch1
recv 3 from ch2
recv 5 from ch1

复制代码

上面是多路复用 select 语句的读通道形式，下面是它的写通道形式，只要有一个通道能写进去，它就会打破阻塞。

select {
  case ch1 <- v:
      fmt.Println("send to ch1")
  case ch2 <- v:
      fmt.Println("send to ch2")
}
复制代码

非阻塞读写

前面咱们讲的读写都是阻塞读写，Go 语言还提供了通道的非阻塞读写。当通道空时，读操做不会阻塞，当通道满时，写操做也不会阻塞。非阻塞读写须要依靠 select 语句的 default 分支。当 select 语句全部通道都不可读写时，若是定义了 default 分支，那就会执行 default 分支逻辑，这样就起到了不阻塞的效果。下面咱们演示一个单生产者多消费者的场景。生产者同时向两个通道写数据，写不进去就丢弃。

package main

import "fmt"
import "time"

func send(ch1 chan int, ch2 chan int) {
	i := 0
	for {
		i++
		select {
			case ch1 <- i:
				fmt.Printf("send ch1 %d\n", i)
			case ch2 <- i:
				fmt.Printf("send ch2 %d\n", i)
			default:
		}
	}
}

func recv(ch chan int, gap time.Duration, name string) {
	for v := range ch {
		fmt.Printf("receive %s %d\n", name, v)
		time.Sleep(gap)
	}
}

func main() {
        // 无缓冲通道
	var ch1 = make(chan int)
	var ch2 = make(chan int)
	// 两个消费者的休眠时间不同，名称不同
	go recv(ch1, time.Second, "ch1")
	go recv(ch2, 2 * time.Second, "ch2")
	send(ch1, ch2)
}

------------
send ch1 27
send ch2 28
receive ch1 27
receive ch2 28
send ch1 6708984
receive ch1 6708984
send ch2 13347544
send ch1 13347775
receive ch2 13347544
receive ch1 13347775
send ch1 20101642
receive ch1 20101642
send ch2 26775795
receive ch2 26775795
...
复制代码

从输出中能够明显看出有不少的数据都丢弃了，消费者读到的数据是不连续的。若是将 select 语句里面的 default 分支干掉，再运行一次，结果以下

send ch2 1
send ch1 2
receive ch1 2
receive ch2 1
receive ch1 3
send ch1 3
receive ch2 4
send ch2 4
send ch1 5
receive ch1 5
receive ch1 6
send ch1 6
receive ch1 7
复制代码

能够看到消费者读到的数据都连续了，可是每一个数据只给了一个消费者。select 语句的 default 分支很是关键，它是决定通道读写操做阻塞与否的关键。

Java 也有通道

通道在其它语言里面的表现形式是队列，在 Java 语言里，带缓冲通道就是并发包内置的 java.util.concurrent.ArrayBlockingQueue，无缓冲通道也是并发包内置的 java.util.concurrent.SynchronousQueue。ArrayBlockingQueue 的内部实现形式是一个数组，多线程读写时须要使用锁来控制并发访问。不过像 Go 语言提供的多路复用效果，Java 语言就没有内置的实现了。

通道内部结构

Go 语言的通道内部结构是一个循环数组，经过读写偏移量来控制元素发送和接受。它为了保证线程安全，内部会有一个全局锁来控制并发。对于发送和接受操做都会有一个队列来容纳处于阻塞状态的协程。

type hchan struct {
  qcount uint  // 通道有效元素个数
  dataqsize uint   // 通道容量，循环数组总长度
  buf unsafe.Pointer // 数组地址
  elemsize uint16 // 内部元素的大小
  closed uint32 // 是否已关闭 0或者1
  elemtype *_type // 内部元素类型信息
  sendx uint // 循环数组的写偏移量
  recvx uint // 循环数组的读偏移量
  recvq waitq // 阻塞在读操做上的协程队列
  sendq waitq // 阻塞在写操做上的协程队列
  
  lock mutex // 全局锁
}
复制代码

这个循环队列和 Java 语言内置的 ArrayBlockingQueue 结构一模一样。从这个数据结构中咱们也能够得出结论：队列在本质上是使用共享变量加锁的方式来实现的，共享变量才是并行交流的本质。

class ArrayBlockingQueue extends AbstractQueue {
  Object[] items;
  int takeIndex;
  int putIndex;
  int count;
  ReentrantLock lock;
  ...
}
复制代码

因此读者不要认为 Go 语言的通道很神奇，Go 语言只是对通道设计了一套便于使用的语法糖，让这套数据结构显的平易近人。它在内部实现上和其它语言的并发队列大同小异。

阅读《快学 Go 语言》更多章节，长按图片识别二维码关注公众号「码洞」