Golang的channel
目录
基本使用
channel 数据结构
阻塞的协程队列
协程节点
构建 channel
写流程
读流程
非阻塞与阻塞
closechan(关闭)
基本使用
创建无缓存 channel
c := make(chan int) //创建无缓冲的通道 cc := make(chan int,0) //创建无缓冲的通道 c创建有缓存 channel
c := make(chan int, 3) //创建无缓冲的通道 c
例子:
package mainimport ("fmt""time"
)func main() {c := make(chan int, 3) //创建有缓冲的通道 c//内置函数 len 返回未被读取的缓冲元素数量，cap 返回缓冲区大小fmt.Printf("len(c)=%d, cap(c)=%d\n", len(c), cap(c))go func() {defer fmt.Println("子go程结束")for i := 0; i < 3; i++ {c <- ifmt.Printf("子go程正在运行[%d]: len(c)=%d, cap(c)=%d\n", i, len(c), cap(c))}}()time.Sleep(2 * time.Second) //延时2sfor i := 0; i < 3; i++ {num := <-c //从c中接收数据，并赋值给numfmt.Println("num = ", num)}fmt.Println("main进程结束")
}
channel 数据结构
type hchan struct {qcount uint // total data in the queuedataqsiz uint // size of the circular queuebuf unsafe.Pointer // points to an array of dataqsiz elementselemsize uint16closed uint32elemtype *_type // element typesendx uint // send indexrecvx uint // receive indexrecvq waitq // list of recv waiterssendq waitq // list of send waiterslock mutex
}
hchan：channel 数据结构
• qcount：当前 channel 中存在多少个元素；
• dataqsize: 当前 channel 能存放的元素容量；
• buf：channel 中用于存放元素的环形缓冲区；
• elemsize：channel 元素类型的大小；
• closed：标识 channel 是否关闭；
• elemtype：channel 元素类型；
• sendx：发送元素进入环形缓冲区的 index；
• recvx：接收元素所处的环形缓冲区的 index；
• recvq：因接收而陷入阻塞的协程队列；
• sendq：因发送而陷入阻塞的协程队列；
lock mutex 锁
阻塞的协程队列
type waitq struct {first *sudoglast *sudog
}waitq：阻塞的协程队列
• first：队列头部
• last：队列尾部
协程节点
sudog：用于包装协程的节点
type sudog struct {g *gnext *sudogprev *sudogelem unsafe.Pointer // data element (may point to stack)isSelect boolc *hchan
}
• g：goroutine，协程；
• next：队列中的下一个节点；
• prev：队列中的前一个节点；
• elem: 读取/写入 channel 的数据的容器;
• isSelect：标识当前协程是否处在 select 多路复用的流程中；
• c：标识与当前 sudog 交互的 chan.
构建 channel
func makechan(t *chantype, size int) *hchan {elem := t.elem// ...mem, overflow := math.MulUintptr(elem.size, uintptr(size))if overflow || mem > maxAlloc-hchanSize || size < 0 {panic(plainError("makechan: size out of range"))}var c *hchanswitch {case mem == 0:// Queue or element size is zero.c = (*hchan)(mallocgc(hchanSize, nil, true))// Race detector uses this location for synchronization.c.buf = c.raceaddr()case elem.ptrdata == 0:// Elements do not contain pointers.// Allocate hchan and buf in one call.c = (*hchan)(mallocgc(hchanSize+mem, nil, true))c.buf = add(unsafe.Pointer(c), hchanSize)default:// Elements contain pointers.c = new(hchan)c.buf = mallocgc(mem, elem, true)}c.elemsize = uint16(elem.size)c.elemtype = elemc.dataqsiz = uint(size)lockInit(&c.lock, lockRankHchan)return
}
写流程
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {// ...//加锁lock(&c.lock)// ...//写时存在阻塞读协程if sg := c.recvq.dequeue(); sg != nil {// Found a waiting receiver. We pass the value we want to send// directly to the receiver, bypassing the channel buffer (if any).send(c, sg, ep, func() { unlock(&c.lock) }, 3)return true}//写时不存在阻塞读协程,且缓冲区不满仍有空间if c.qcount < c.dataqsiz {// Space is available in the channel buffer. Enqueue the element to send.qp := chanbuf(c, c.sendx)typedmemmove(c.elemtype, qp, ep)c.sendx++if c.sendx == c.dataqsiz {c.sendx = 0}c.qcount++unlock(&c.lock)return true}//写时不存在阻塞读协程,且缓冲区满了没有空间// ...gp := getg()mysg := acquireSudog()mysg.elem = epmysg.g = gpmysg.c = cgp.waiting = mysgc.sendq.enqueue(mysg)atomic.Store8(&gp.parkingOnChan, 1)gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)gp.waiting = nilclosed := !mysg.successgp.param = nilmysg.c = nilreleaseSudog(mysg)return true
}
总结:
1.当写时存在阻塞读协程,我们直接用
2.当写时不存在阻塞读协程,且缓冲区不满仍有空间时,我们直接加入环形缓冲区中
3.当写时不存在阻塞读协程,且缓冲区满了没用空间时,加入阻塞写协程队列中
注意:
1.有阻塞读协程和缓冲区满之间只有一个条件符合
2.对于未初始化的 chan，写入操作会引发死锁
3.对于已关闭的 chan，写入操作会引发 panic.
读流程
读流程与写流程差不多,不同点:
1.加入的是阻塞读队列
2.当环形缓冲区有和无数据时会有不同的操作
注意:
1.读空channel, park挂起,引起死锁
2.channel 已关闭且内部无元素,直接解锁返回
非阻塞与阻塞
区别:
非阻塞模式下，读/写 channel 方法通过一个 bool 型的响应参数，用以标识是否读取/写入成功.
• 所有需要使得当前 goroutine 被挂起的操作，在非阻塞模式下都会返回 false；
• 所有是的当前 goroutine 会进入死锁的操作，在非阻塞模式下都会返回 false；
• 所有能立即完成读取/写入操作的条件下，非阻塞模式下会返回 true.
何时进入非阻塞
默认情况下，读/写 channel 都是阻塞模式，只有在 select 语句组成的多路复用分支中，
与 channel 的交互会变成非阻塞模式：
在sudog：用于包装协程的节点
• isSelect：标识当前协程是否处在 select 多路复用的流程中；
closechan(关闭)
func closechan(c *hchan) {if c == nil {panic(plainError("close of nil channel"))}lock(&c.lock)if c.closed != 0 {unlock(&c.lock)panic(plainError("close of closed channel"))}c.closed = 1var glist gList// release all readersfor {sg := c.recvq.dequeue()if sg == nil {break}if sg.elem != nil {typedmemclr(c.elemtype, sg.elem)sg.elem = nil}gp := sg.ggp.param = unsafe.Pointer(sg)sg.success = falseglist.push(gp)}// release all writers (they will panic)for {sg := c.sendq.dequeue()if sg == nil {break}sg.elem = nilgp := sg.ggp.param = unsafe.Pointer(sg)sg.success = falseglist.push(gp)}unlock(&c.lock)// Ready all Gs now that we've dropped the channel lock.for !glist.empty() {gp := glist.pop()gp.schedlink = 0goready(gp, 3)关闭未初始化过的 channel 会 panic；
• 加锁；
• 重复关闭 channel 会 panic；
• 将阻塞读协程队列中的协程节点统一添加到 glist；
• 将阻塞写协程队列中的协程节点统一添加到 glist；
• 唤醒 glist 当中的所有协程.
防止还有协程挂起,没有被唤醒的资源浪费
参考:小徐先生1212 -- Golang Channel实现原理