16 协程间通信

[TOC]

协程间通信

通道 chan

channel通道是CSP派生的同步原语之一，虽然它可以用来同步内存访问，但它一般用于协程之间传递信息。

• 如Context中就是封装了通道，主要用于协程退出通知。

• 就像水管一样，chan 通道充当着传递信息的角色，数据可以沿着通道道传递，在管道另一头读取数据。

通道的结构是环形队列

• 队列的特点，先进先出，因此通道中的数据是先进先出的。

• 有缓存的通道，通道中存储的数据是线性数组，但是用数组和序号模拟了一个环形队列。

CSP 通信顺序进程

CSP （Communicating Sequential Processes 通信顺序进程）是用于描述并发系统中交互模式的形式化语言，它通过通道传递消息。

• 以往多进程或多线程程序通常采用共享内存进行交流，通过信号量等手段实现同步机制。CSP的思想是Go语言并发的重要设计思想，通过命名的通道发送或接收值进行通信。

• 在最初的设计中，通道是无缓冲的，因此发送操作会阻塞，直到被接收端接收后才能继续发送，从而提供了一种同步机制。

通道的基本使用

通道可以是有缓冲区也可以无缓冲区

• make(chan type) 创建无缓冲区通道，无缓冲区通道也称为同步通道

• make(chan type, 缓冲区大小 int) 创建有缓冲区通道

• 读取通道数据 r := <- chan

• 向通道写入数据 chan <- data

通道的内置函数

• close(ch) 关闭通道

• len(ch) int 通道中的数据数量int

• cap(ch) int 通道的容量，即缓冲区容量

• 判断通道是否close()，可以读取通道数据第二个返回值 【 data,isOpen := <- chan 】通道开启则 isOpen 为 true，isOpen 为 false则说明通道close。

无缓冲通道基本案例：

• 无缓冲通道读写数据在不能读写时都是阻塞的，已经写入的数据未被读取走则写入端会阻塞，通道中无数据可读则读端会阻塞等待。

• 案例中main主协程不断读取通道中的数据，当读取到数据为1时退出main主协程。

package main

import (
	"fmt"
	"time"
)

func main() {
	ch := make(chan int)
	go func() {
		time.Sleep(1*time.Second)
		ch <- 0
		ch <- 1
	}()
	beg := time.Now().UnixNano() / 1e6
	for {
		data := <- ch
		fmt.Printf("通道中读取到数据%d \n",data)
		if 1 == data {
			fmt.Printf("退出main 主协程，耗时%d毫秒\n",time.Now().UnixNano() / 1e6 - beg)
			return
		}
	}
	// 输出结果：
	//通道中读取到数据0
	//通道中读取到数据1
	//退出main 主协程，耗时1009毫秒
}

有缓冲区通道案例：

• 向有缓冲区通道写入数据，缓冲区满后会阻塞，直到通道缓冲区数据被取走。

• 读取有缓冲区通道的数据，缓冲区中没有数据则会阻塞等待。

func main() {
	ch := make(chan string,3)
	go func() {
		ch <- "数据0"
		ch <- "数据1"
		ch <- "数据2"
		fmt.Println("协程已经向通道写入3个数据，缓冲区满了再写就阻塞")
		ch <- "EXIT"
		close(ch)
	}()
	beg := time.Now().UnixNano() / 1e6
	time.Sleep(2 * time.Second)
	for {
		data := <- ch
		fmt.Printf("通道中读取到数据：%s \n",data)
		if "EXIT" == data {
			fmt.Printf("退出main 主协程，耗时%d毫秒\n",time.Now().UnixNano() / 1e6 - beg)
			return
		}
	}
	//输出结果：
	//协程已经向通道写入3个数据，缓冲区满了再写就阻塞
	//通道中读取到数据：数据0
	//通道中读取到数据：数据1
	//通道中读取到数据：数据2
	//通道中读取到数据：EXIT
	//退出main 主协程，耗时2009毫秒
}

单方向通道

默认的chan都是双向的，方向只是chan一种属性，在作为只读或者只写函数参数时， 可以有效控制channel使用范围

• 双向chan可以隐式转换为任意单向，但是单向不得转换其它。

• 双向转只写单向 var w chan <- int = ch

• 双向转只读单向 var r <- chan int = ch

单方向通道案例：

• write() 函数接收通道参数【ch chan<- int】只写通道。

• read() 函数接收通道参数【ch <-chan int】只读通道。

• 当 write() 函数 defer close(ch) 通道关闭，read() 函数中通道会不断读取到零值，因此关闭的通道并不会读取不到数据而是不断读取到零值造成死循环，通过【data,isOpen := <- ch】中isOpen值判断通道是否关闭，关闭则结束读取，不会造成死循环。

package main

import "fmt"

func write(ch chan<- int, closeCh chan<- string) {
	defer close(ch)	// 关闭通道
	for i:=1;i<=5;i++ {
		ch <- i
	}
	closeCh <- "write close"
}

func read(ch <-chan int, closeCh chan<- string){
	for {
		data,isOpen := <- ch
		fmt.Printf("read 读取到管道数据%d，通道开启=%v \n",data,isOpen)
		if !isOpen {
			break
		}
	}
	closeCh <- "read close"
}

func main() {
	ch := make(chan int,2)
	closeCh := make(chan string)

	go write(ch,closeCh)
	go read(ch,closeCh)

	count :=0
	for {
		str := <- closeCh
		fmt.Printf("main主协程收到closeCh数据=%s \n",str)
		count++
		// 收到2次 closeCh通道的数据后说明write()、read()都结束
		if 2 == count {
			close(closeCh)
			return
		}
	}
	// 输出结果：
	//read 读取到管道数据1，通道开启=true
	//read 读取到管道数据2，通道开启=true
	//read 读取到管道数据3，通道开启=true
	//read 读取到管道数据4，通道开启=true
	//main主协程收到closeCh数据=write close
	//read 读取到管道数据5，通道开启=true
	//read 读取到管道数据0，通道开启=false
	//main主协程收到closeCh数据=read close
}

select 多路复用

select 是类UNIX系统提供的一个多路复用系统API，Go借用多路复用的概念，提供了select关键字，用于多路监听多个通道。

• 使用通道时往往会与select结合，因为经常有多个通道与多个协程进行通信的情况，为了避免一个通道的读写陷入堵塞，影响其它通道的正常读写。

• 当监听的通道没有状态是可读或可写的，select是阻塞的。

• 只要监听的通道中有一个状态是可读或可写的，则select就不会阻塞，而是进入处理就绪通道的分支流程。

• 如果监听的通道有多个可读或可写的状态，则select随机选取一个处理。

select 随机选择机制

当多个通道同时准备执行读写操作时，select会随机选择一个执行。

• select 中没有任何准备好的通道，select会陷入阻塞，直到一个通道准备好。

• 如果 select 中有default分支，则通道陷入阻塞时会执行default分支。

• 如果不设置default分支，也可以设置一个超时器，使用time.After() 设置一个超时器。

select 与循环配合使用

select本身没有循环能力，select会执行一次就退出，但实际使用中往往是不断监听，因此要使用循环配合使用。

for {
    select {
        case <- a:
            ...
        case <- b:
            ...
    }
}

select 使用案例

select 多路复用

• 3个通道都有数据，select是随机选择读取。

• time.Tick 定时器，到定时时间 Tick 通道就会写入数据，select监听到则执行超时退出监听。

package main

import (
	"fmt"
	"time"
)

func write(ch chan<- string,goName string){
	for i:=0;i<3;i++ {
		ch <- fmt.Sprintf("协程%s写入数据%d",goName,i)
	}
}
func main() {
	ch1 := make(chan string,2)
	ch2 := make(chan string,2)
	ch3 := make(chan string,2)
	go write(ch1,"协程1")
	go write(ch2,"协程2")
	go write(ch3,"协程3")
	tick := time.Tick(1000 * time.Millisecond)
	lab:	// break 标签
	for {
		select {
			case data := <- ch1:
				fmt.Println(data)
			case data := <- ch2:
				fmt.Println(data)
			case data := <- ch3:
				fmt.Println(data)
			case <- tick:
				fmt.Println("时间到 退出监听")
				break lab
			default:
				fmt.Println("default 语句")
				time.Sleep(100 * time.Millisecond)
		}
	}
//	输出结果：
//	default 语句
//	协程协程3写入数据0
//	协程协程1写入数据0
//	协程协程1写入数据1
//	协程协程2写入数据0
//	协程协程2写入数据1
//	协程协程2写入数据2
//	协程协程3写入数据1
//	协程协程1写入数据2
//	协程协程3写入数据2
//	default 语句 ...
//	时间到 退出监听
}

退出通知机制下的 select 陷阱

读取已经关闭的通道不会引起阻塞，也不会导致panic，而是立即返回该通道存储类型的零值。

• 关闭select监听的某个通道能让select立即感知这种通知，然后进行相应的处理，这就是所谓退出通知机制。context标准库就是利用这种机制处理更复杂的通知机制。

select 陷阱案例：

• 案例中 defer close(ch) 关闭了通道，但是关闭的通道不代表是读取不到数据，关闭的通道会不断读取到通道类型的零值，在读取得到通道零值的情况下甚至都不会执行default分支，案例中设置了Tick超时退出，否则会进入无限循环。

• 结束读取零值的情况可以使用让通道赋值为nil，因此要结束这种情况只能是读取通道的第二个参数 isOpen ，如果isOpen为false就将通道赋值为nil。

package main

import (
	"fmt"
	"time"
)

func write(ch chan<- string,goName string){
	defer close(ch)
	for i:=1;i<=3;i++ {
		ch <- fmt.Sprintf("协程%s写入数据%d",goName,i)
	}
}
func main() {
	ch1 := make(chan string,2)
	ch2 := make(chan string,2)
	ch3 := make(chan string,2)
	go write(ch1,"协程1")
	go write(ch2,"协程2")
	go write(ch3,"协程3")
	tick := time.Tick(1000 * time.Millisecond)
lab:	// break 标签
	for {
		select {
		case data,isOpen := <- ch1:
			fmt.Printf("读取数据：%s\n",data)
			if !isOpen {
				ch1 = nil // 通道置为nil时，就不会无限读取通道零值。
			}
		case data,isOpen := <- ch2:
			fmt.Printf("读取数据：%s\n",data)
			if !isOpen {
				ch2 = nil
			}
		case data,isOpen := <- ch3:
			fmt.Printf("读取数据：%s\n",data)
			if !isOpen {
				ch3 = nil
			}
		case <- tick:
			fmt.Println("时间到 退出监听")
			break lab
		default:
			fmt.Println("default 语句")
			time.Sleep(100 * time.Millisecond)
		}
	}
	//	输出结果：
	//default 语句
	//读取数据：协程协程2写入数据1
	//读取数据：协程协程1写入数据1
	//读取数据：协程协程1写入数据2
	//读取数据：协程协程2写入数据2
	//读取数据：协程协程1写入数据3
	//读取数据：协程协程3写入数据1
	//读取数据：协程协程3写入数据2
	//读取数据：协程协程3写入数据3
	//读取数据：协程协程2写入数据3
	//读取数据：
	//default 语句
	//读取数据：
	//读取数据：
	//default 语句 ...
	//时间到 退出监听
}

Context 标准库

Go 1.7 提供了context 标准库，用于解决复杂的并发结构处理。

• 实际编程中协程（goroutine）会启动新的协程，新的协程又会开启新的协程，最终形成一个类似于树形的结构，由于Go的协程没有父子概念，这个协程的树形结构只是开发者抽象所得，Go并没有维护一个协程的树形结构。

• context 提供了协程退出通知和元数据传递的功能，context会跟踪协程调用，在其内部维护一个调用树，并在这些调用树中传递通知和元数据。

• context 作为协程执行的函数的第一个参数，每开启新协程进行一次包装，形成一个调用树。

context 跟踪协程调用树，并在这些协程调用树中传递通知和元数据。

• 退出通知机制，通知可以传递给整个协程调用树上的每一个协程。

• 传递数据，数据可以传递给整个协程调用树上的每一个协程。

context 整体工作机制

第一个创建context的协程是root节点。

• root节点负责创建一个实现context接口的具体对象，而后将对象作为参数传递到新启动的协程，下游的协程可以继续封装该对象，再传递到更下游的协程。

• context 对象在传递过程中最终形成一个树状的数据结构，通过root节点的context对象就能遍历整个context对象树，通知和消息就可以通过root节点传递出去，实现上游协程对下游的消息传递。

context 接口

context是接口，标准库提供了一些实现Context接口的标准context。

• func Background() Context 函数，func TODO() Context 函数，用于构造Context树的根节点对象。

//	Deadline context实现了超时控制，否则方法返回ok=true，deadline是超时时间
func Deadline() (deadline time.Time, ok bool)
//	Done 后端被调用的协程应该监听该方法返回的chan，以便及时释放资源
func Done() <-chan struct{} 
//	Err Done返回的chan收到通知的时候，才可以访问Err()获知因为什么原因被取消
func Err() error
//	Value 可以访问上游协程传递给下游协程的值
func Value(key interface{}) interface{} 

context包下，With***() 包装函数用来构建不同功能的Context具体对象

• 如：func WithCancel(parent Context) (ctx Context,cancel CancelFunc) ，用于创建一个带有退出通知的Context具体对象，内部创建一个cancelCtx的类型实例。

• 此外还有创建带超时通知的Context具体对象 WithDeadline() ，WithTimeout() 。

• 创建能够传递数据的Context具体对象，WithValue()。

canceler 接口（context的扩展接口）

canceler接口是一个扩展接口，取消通知的Context具体类型需要实现此接口，context包中的cancelCtx和timerCtx都实现了该接口。

type canceler interface {
	cancel(removeFromParent bool, err error)
	Done() <-chan struct{}
}

context 实现者 emptyCtx

emptyCtx实现了context接口，但不具备任何功能，因为它所有的方法实现都是空实现，它存在的目的是作为Context树的root节点，因为context包的使用思路就是不停地调用context包提供的包装函数来创建具有特殊功能的Context实例，每一个Context实例的创建都以上一个Context对象作为参数，最终形成一个树状结构。

• 而实际上，Background()和TODO()函数构造的Context就是emptyCtx的实例，所以往往构造出作为root节点。

context 实现者 cancelCtx

cancelCtx 实现了Context接口和canceler接口，具有退出通知方法。

• 退出通知机制不仅能通知自己，也可以通知下游节点。

context 实现者 timerCtx

timerCtx 实现了Context接口，内部封装了cancelCtx的实例，同时有一个deadline变量，用来实现定时退出通知。

context 实现者 valueCtx

valueCtx实现了Context接口，内部封装Context接口类型，同时封装了k/v的存储变量。

• valueCtx可以用来传递通知消息。

context Done()的基本用法

让协程退出就是监听Done()中的通道，通道得到一个数据就是退出协程的信号，实际上收到这个信号协程要做什么是由通道监听方决定的。

context.WithCancel() 协程退出通知案例：

• WithCancel()包装的context中，监听Done()方法的通道，以确定是否收到退出的信号。

• 当调用CancelFunc func() 时，就是向管道写一个数据，这个行为就是发送退出信号。

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

func work(ctx context.Context,name string){
	defer wg.Done()
	for {
		select {
			case <- ctx.Done():
				fmt.Printf("协程%s 收到退出的信号，结束函数。\n",name)
				return // 收到退出信号，结束这个函数
			default:
				fmt.Printf("协程%s 正在运行\n",name)
				time.Sleep(1 * time.Second)
		}
	}
}
var wg sync.WaitGroup
func main() {
    wg.Add(1)
	ctxCancel,cancelFun := context.WithCancel(context.Background())
	go work(ctxCancel,"【WithCancel测试】")
	// 休眠3s后，主动调用cancelFun让协程退出
	time.Sleep(3*time.Second)
	cancelFun()
	wg.Wait()
	fmt.Println("main")
//	输出结果
//	协程【WithCancel测试】 正在运行
//	协程【WithCancel测试】 正在运行
//	协程【WithCancel测试】 正在运行
//	协程【WithCancel测试】 收到退出的信号，结束函数。
//	main
}

context.WithDeadline() 协程超时退出案例：

• 使用context.WithDeadline(context.Background(),timeOut)函数进行包装一个协程超时退出的context，实际上是当达到超时时间后向Done()的通道发送数据，而非WithCancel中手动调用发送数据的方法。

func main() {
	wg.Add(1)
	timeOut := time.Now().Add(3 * time.Second) // 3秒后超时
	ctxDL,_ :=context.WithDeadline(context.Background(),timeOut)
	go work(ctxDL,"【WithDeadline测试】")
	wg.Wait()
	fmt.Println("main")
	//	输出结果
	//协程【WithDeadline测试】 正在运行
	//协程【WithDeadline测试】 正在运行
	//协程【WithDeadline测试】 正在运行
	//协程【WithDeadline测试】 收到退出的信号，结束函数。
	//main
}

context 上游协程控制下游协程退出

main开启work1，work1开启work2，work2开启work3，上游可以控制下游。

• work1() 的协程定时会在5s后发送退出信号。

• work2()的协程是3s后退出。

• work2协程在default分支中控制work3()协程退出。

• context中，上游协程收到Done()退出信号它所开启的下游协程也会收到，因此如果把work1()的延迟退出改成1s，在1s后所有协程都会一起退出。

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

func work1(ctx context.Context,name string){
	defer wg.Done()
	ctx2,_ := context.WithTimeout(ctx,timeOutArray[1] * time.Second)
	go work2(ctx2,"【第二层context】")
	count := 0
	for {
		select {
		case <- ctx.Done():
			fmt.Printf("协程%s 收到退出的信号，结束函数。\n",name)
			return // 收到退出信号，结束这个函数
		default:
			if 3 == count {

			}
			fmt.Printf("协程%s 正在运行\n",name)
			time.Sleep(1 * time.Second)
			count++
		}
	}
}
func work2(ctx context.Context,name string){
	defer wg.Done()
	ctx3,canFun := context.WithCancel(ctx)
	go work3(ctx3,"【第三层context】")
	for {
		select {
		case <- ctx.Done():
			fmt.Printf("协程%s 收到退出的信号，结束函数。\n",name)
			return // 收到退出信号，结束这个函数
		default:
			fmt.Printf("协程%s 正在运行\n",name)
			time.Sleep(1 * time.Second)
			canFun()
		}
	}
}
func work3(ctx context.Context,name string){
	defer wg.Done()
	for {
		select {
		case <- ctx.Done():
			fmt.Printf("协程%s 收到退出的信号，结束函数。\n",name)
			return // 收到退出信号，结束这个函数
		default:
			fmt.Printf("协程%s 正在运行\n",name)
			time.Sleep(1 * time.Second)
		}
	}
}
var wg sync.WaitGroup
var timeOutArray = [3]time.Duration{5,3}
func main() {
	wg.Add(3)
	ctxDL,_ :=context.WithTimeout(context.Background(),timeOutArray[0] * time.Second)
	go work1(ctxDL,"【第一层context】")
	wg.Wait()
	fmt.Println("main")
	//输出结果：
	//协程【第二层context】 正在运行
	//协程【第一层context】 正在运行
	//协程【第三层context】 正在运行
	//协程【第一层context】 正在运行
	//协程【第二层context】 正在运行
	//协程【第三层context】 正在运行
	//协程【第三层context】 收到退出的信号，结束函数。
	//协程【第一层context】 正在运行
	//协程【第二层context】 正在运行
	//协程【第二层context】 收到退出的信号，结束函数。
	//协程【第一层context】 正在运行
	//协程【第一层context】 正在运行
	//协程【第一层context】 收到退出的信号，结束函数。
	//main
}

• 将work1延迟关闭改为1，则work1在1s后退出，work2、work3都会一起退出。

var timeOutArray = [3]time.Duration{1,3}

/*
    协程【第一层context】 正在运行
    协程【第三层context】 正在运行
    协程【第二层context】 正在运行
    协程【第二层context】 正在运行
    协程【第三层context】 收到退出的信号，结束函数。
    协程【第一层context】 收到退出的信号，结束函数。
    协程【第二层context】 收到退出的信号，结束函数。
    main
*/

context 数据传递的争议

context传递数据的争议

• context包主要是解决协程的通知退出，传递数据是其一个额外功能，可以使用它传递一些元信息，但是不要使用context传递影响业务流程的数据。

• context传递数据的缺点：传递的值都是interface{}类型的值编译器无法进行严格类型检查，interface{}数据需要类型断言才能使用，值在传递过程中可能被后续的服务覆盖，传递信息不透明不利于后期维护。

• context适合传递的数据：不影响业务主逻辑的可选数据，如日志信息和调试信息。

• context包提供的核心功能是多个协程间退出通知机制，传递数据只是一个辅助功能。

需要使用context传递数据就要使用context.WithValue对context进行包装

• 主要注意context.WithValue本身包装出的没有超时信号功能，如果直接包装context.Background()将无法设置超时退出。

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

func workValue(ctx context.Context,name string){
	defer wg.Done()
	for {
		select {
			case <- ctx.Done() :
				fmt.Printf("协程%s 收到退出信号\n",name)
				return
			default:
				v,_ := ctx.Value("data").(string)
				fmt.Printf("协程%s 运行中 收到数据：%s \n",name,v)
				time.Sleep(1*time.Second)
		}
	}
}
var wg sync.WaitGroup
func main() {
	// context.WithValue本身没有发生超时信号Done()的功能，因此先WithTimeout()包装了3秒超时的功能，以避免案例陷入死循环
	wg.Add(1)
	ctxTimeOut,_ := context.WithTimeout(context.Background(),3 * time.Second)
	ctxV := context.WithValue(ctxTimeOut,"data","这是数据value ^.^")
	go workValue(ctxV,"【测试WithValue】")
	wg.Wait()
	fmt.Println("main")
//	输出结果
//	协程【测试WithValue】 运行中 收到数据：这是数据value ^.^
//	协程【测试WithValue】 运行中 收到数据：这是数据value ^.^
//	协程【测试WithValue】 运行中 收到数据：这是数据value ^.^
//	协程【测试WithValue】 收到退出信号
//	main
}