channel 是 Golang 提供的 goroutine 间的通信形式,能够让一个 goroutine 发送特定值到另一个 goroutine。
个性
通道没有缓冲区,或者有缓冲区但缓冲区没有数据时,从通道读取数据会阻塞,直到有协程向通道中写入数据。相似地,通道没有缓冲区,或者缓冲区已满时,向通道写入数据也会阻塞,直到有协程从通道读取数据。对于值为 nil 的通道,无论读写都会阻塞,而且是永恒阻塞。
应用内置函数 close 能够敞开通道,尝试向已敞开的通道发送数据会触发 panic,但此时依然可读。通道读取的表达式最多有两个返回值:
x, ok := <-ch
第一个变量示意读出的数据,第二个变量示意是否胜利读取了数据,它的值只跟通道缓冲区中是否有数据无关,与通道的敞开状态无关。
实现原理
数据结构
源码 src/runtime/chan.go:hchan 定义了 channel 的数据结构:
type hchan struct { qcount uint // 以后队列中残余元素个数 dataqsiz uint // 环形队列长度,即能够寄存的元素个数 buf unsafe.Pointer // 环形队列指针 elemsize uint16 // 每个元素的大小 closed uint32 // 标识敞开状态 elemtype *_type // 元素类型 sendx uint // 队列下标,批示元素写入时寄存到队列中的地位 recvx uint // 队列下标,批示元素从队列的该地位读出 recvq waitq // 期待读音讯的 goroutine 队列 sendq waitq // 期待写音讯的 goroutine 队列 lock mutex // 互斥锁,chan 不容许并发读写}
能够看出 channel 由队列、类型信息、goroutine 期待队列组成。
环形队列
channel 外部实现了一个环形队列作为其缓冲区,队列的长度是创立 channel 时指定的。下图展现了一个可缓存 6 个元素的 channel 示意图:
- dataqsiz 表明了队列长度为6,即可缓存6个元素;
- buf 指向队列的内存地址;
- qcount 示意队列中还有两个元素;
- sendx 示意后续写入的数据存储的地位,取值为 [0, 6);
- recvx 示意读取数据的地位, 取值为[0, 6)。
类型信息
一个 channel 只能传递一种类型的值:
- elemtype 代表类型,用于数据传递过程中的赋值;
- elemsize 代表类型大小,用于在buf中定位元素地位。
期待队列
从 channel 读取数据时,如果没有缓冲区或者缓冲区为空,则以后协程会被阻塞,并被退出 recvq 队列。向 channel 写入数据时,如果没有缓冲区或者缓冲区已满,则以后协程同样会被阻塞,而后退出到 sendq 的队列。处于期待队列中的协程会在其余协程操作 channel 时被唤醒。
下图展现了一个没有缓冲区的 channel,并有几个协程正在阻塞期待读取数据:
相干操作
创立通道
创立 channel 的过程实际上就是初始化 hchan 构造,类型信息和缓冲区长度由 make 语句传入,buf 的大小则由元素大小和缓冲区长度独特决定。
源码 src/runtime/chan.go 中定义了创立 channel 的函数 makechan(),精简版的代码如下所示:
func makechan(t *chantype, size int) *hchan { mem, overflow := math.MulUintptr(elem.size, uintptr(size)) var c *hchan switch { case mem == 0: c = (*hchan)(mallocgc(hchanSize, nil, true)) c.buf = c.raceaddr() case elem.ptrdata == 0: c = (*hchan)(mallocgc(hchanSize+mem, nil, true)) c.buf = add(unsafe.Pointer(c), hchanSize) default: c = new(hchan) c.buf = mallocgc(mem, elem, true) } c.elemsize = uint16(elem.size) c.elemtype = elem c.dataqsiz = uint(size) return c}
发送数据
发送数据的操作最终都转化成了 chansend() 函数,次要代码和逻辑如下:
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool { // 如果通道为 nil,非阻塞式发送的话间接返回 false,否则将以后协程挂起 if c == nil { if !block { return false } gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2) throw("unreachable") } // 对于非阻塞式发送,如果通道未敞开且没有缓冲空间的话,间接返回 false if !block && c.closed == 0 && full(c) { return false } // 加锁,并发平安 lock(&c.lock) // 如果通道敞开了,间接 panic if c.closed != 0 { unlock(&c.lock) panic(plainError("send on closed channel")) } // 如果接管队列不为空,间接将要发送的数据发送到队首的 goroutine if sg := c.recvq.dequeue(); sg != nil { send(c, sg, ep, func() { unlock(&c.lock) }, 3) return true } // 对于缓冲区还有闲暇的 channel,拷贝数据到缓冲区,保护相干信息 if c.qcount < c.dataqsiz { qp := chanbuf(c, c.sendx) if raceenabled { raceacquire(qp) racerelease(qp) } typedmemmove(c.elemtype, qp, ep) c.sendx++ if c.sendx == c.dataqsiz { c.sendx = 0 } c.qcount++ unlock(&c.lock) return true } // 没有缓冲空间时,发送方会挂起,并依据以后 goroutine 结构一个 sudog 构造体增加到 sendq 队列中 gp := getg() mysg := acquireSudog() mysg.releasetime = 0 if t0 != 0 { mysg.releasetime = -1 } mysg.elem = ep mysg.waitlink = nil mysg.g = gp mysg.isSelect = false mysg.c = c gp.waiting = mysg gp.param = nil c.sendq.enqueue(mysg) gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2) // 省略被唤醒时局部代码 return true}
读取数据
读取数据的操作最终是转化成了 chanrecv() 函数,次要逻辑如下:
// selected 和 received 返回值别离代表是否可被 select 语句命中以及是否读取到了数据func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) { // 如果 channel 为 nil,非阻塞式读取间接返回,否则间接挂起 if c == nil { if !block { return } gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2) throw("unreachable") } // 非阻塞模式并且没有音讯可读(没有缓冲区或者缓冲区为空),如果 channel 未敞开间接返回 if !block && empty(c) { if atomic.Load(&c.closed) == 0 { return } if empty(c) { if raceenabled { raceacquire(c.raceaddr()) } if ep != nil { typedmemclr(c.elemtype, ep) } return true, false } } // 加锁 lock(&c.lock) // channel 已敞开并且没有音讯可读(没有缓冲区或者缓冲区为空),会接管到零值,typedmemclr 会依据类型清理相应地址的内存 if c.closed != 0 && c.qcount == 0 { if raceenabled { raceacquire(c.raceaddr()) } unlock(&c.lock) if ep != nil { typedmemclr(c.elemtype, ep) } return true, false } // 期待发送队列不为空,如果是非缓冲型 channel,间接拷贝发送者的数据,否则接管队首的数据,并将发送者的数据挪动到环形队列尾部 if sg := c.sendq.dequeue(); sg != nil { recv(c, sg, ep, func() { unlock(&c.lock) }, 3) return true, true } // 缓冲型 channel,buf 里有元素,能够失常接管 if c.qcount > 0 { // Receive directly from queue qp := chanbuf(c, c.recvx) if raceenabled { raceacquire(qp) racerelease(qp) } if ep != nil { typedmemmove(c.elemtype, ep, qp) } typedmemclr(c.elemtype, qp) c.recvx++ if c.recvx == c.dataqsiz { c.recvx = 0 } c.qcount-- unlock(&c.lock) return true, true } // 被阻塞的状况,结构一个 sudog 构造体,保留到 channel 的期待接管队列,并将以后 goroutine 挂起 gp := getg() mysg := acquireSudog() mysg.releasetime = 0 if t0 != 0 { mysg.releasetime = -1 } mysg.elem = ep mysg.waitlink = nil gp.waiting = mysg mysg.g = gp mysg.isSelect = false mysg.c = c gp.param = nil c.recvq.enqueue(mysg) atomic.Store8(&gp.parkingOnChan, 1) gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2) // 省略被唤醒时局部代码 return true, !closed}
敞开通道
敞开某个 channel,最终会执行函数 closechan(),外围代码如下:
func closechan(c *hchan) { // 如果 channel 为 nil,间接 panic if c == nil { panic(plainError("close of nil channel")) } // 加锁,如果 channel 已敞开,间接 panic lock(&c.lock) if c.closed != 0 { unlock(&c.lock) panic(plainError("close of closed channel")) } c.closed = 1 var glist gList // 开释期待接管队列中,向须要返回值的接收者返回相应的零值 for { sg := c.recvq.dequeue() if sg == nil { break } if sg.elem != nil { typedmemclr(c.elemtype, sg.elem) sg.elem = nil } if sg.releasetime != 0 { sg.releasetime = cputicks() } gp := sg.g gp.param = nil if raceenabled { raceacquireg(gp, c.raceaddr()) } glist.push(gp) } // 开释期待发送队列,相干的 goroutine 会触发panic for { sg := c.sendq.dequeue() if sg == nil { break } sg.elem = nil if sg.releasetime != 0 { sg.releasetime = cputicks() } gp := sg.g gp.param = nil if raceenabled { raceacquireg(gp, c.raceaddr()) } glist.push(gp) } unlock(&c.lock) // ...}
常见利用
定时工作
这种用法须要与 timer 联合,分为两种:超时管制和定时执行。
如果须要执行某项操作,但又不想它消耗太长时间,想给它一个超时限度,能够这么做:
select { case <-time.After(100 * time.Millisecond): case <-s.stopc: return false}
期待 100 ms 后,如果 s.stopc 还没有读出数据或者被敞开,就间接完结。
定时执行某个工作也比较简单,例如每隔 1 秒种,执行一次定时工作:
func worker() { ticker := time.Tick(1 * time.Second) for { select { case <- ticker: // 执行工作 } }}
解耦生产者与消费者
应用一个 channel 保留工作,启动 n 个 goroutine 作为工作协程池,这些协程工作在一个有限循环里,从该 channel 读取工作并执行:
func main() { taskCh := make(chan int, 100) go worker(taskCh) for i := 0; i < 10; i++ { taskCh <- i } select { case <-time.After(time.Hour): }}func worker(taskCh <-chan int) { const N = 5 for i := 0; i < N; i++ { go func(id int) { for { task := <- taskCh fmt.Printf("finish task: %d by worker %d\n", task, id) time.Sleep(time.Second) } }(i) }}
管制并发数
有时须要定时执行几百个工作,然而并发数又不能太高,这时就能够通过 channel 来管制并发数。比方上面的例子:
var limit = make(chan int, 3)func main() { // ………… for _, w := range work { go func() { limit <- 1 w() <-limit }() } // …………}
构建一个容量为 3 的 channel,遍历工作列表,每个工作启动一个 goroutine,真正执行工作的动作在 w() 中实现。在执行 w() 之前,先要从 limit 中拿“许可证”,拿到许可证之后,能力执行 w(),并且在执行完工作,要将“许可证”偿还。要留神的是,如果 w() 产生 panic,那“许可证”可能就还不回去了,因而须要应用 defer 来保障。