Go的channel是Go的一大特性，相对于其他编程而言，Go的channel提供了进程内消息通信，数据同步的重要功能，因为是内置数据结构，因此使用起来非常方便，因此，针对这个常用的数据结构，我们需要探究下其实现原理以及设计思想。

https://github.com/golang/go/blob/go1.12.7/src/runtime/chan.go

Go管理channel结构体如下，我们使用make创建channel时，就是初始化hchan结构体，hchan的定义如下：

  2. type hchan struct {
  3.     qcount   uint           // 当前环形队列内的数据个数
  4.     dataqsiz uint           // 环形队列总长度
  5.     buf      unsafe.Pointer // 数据缓冲区指针，指向一个dataqsiz大小的数组，环形队列由该数组实现
  6.     elemsize uint16        // 单个数据的大小
  7.     closed   uint32        // channel是否关闭
  8.     elemtype *_type // element type  // 环形队列内的数据类型
  9.     sendx    uint   // send index  // 队列下标，元素写入队列时的队列的位置，即队列尾
  10.     recvx    uint   // receive index   // 队列下标，下一个被读取的元素在队列的位置，即队列头
  11.     recvq    waitq  // 等待读消息的goroutine队列，环形队列空时，recvq会堆积
  12.     sendq    waitq  // 等待写消息的goroutine队列，环形队列满时，sendq会堆积
  14.     lock mutex // 互斥锁，保证channel并发读写安全
  15. }
  17. type waitq struct {
  18.     first *sudog  // 队列头指针
  19.     last  *sudog  // 队列尾指针
  20. }

由channel的结构体可以看出，组合成channel的几大部分是：

  1. 环形队列，用于缓存数据
  2. groutine等待队列
  3. 互斥锁

channel的创建

我们使用

  1. c := make(chan int, 6)

创建了一个channel,其缓冲区大小为6，元素类型为int。我们通过源码makeChan可以看出，我们调用make创建channel时返回的返回值的类型值hchan指针。

  1. func makechan(t *chantype, size int) *hchan {
  2.     elem := t.elem
  4.     // 单个元素size过大
  5.     if elem.size >= 1<<16 {
  6.         throw("makechan: invalid channel element type")
  7.     }
  8.     // 字节对齐
  9.     if hchanSize%maxAlign != 0 || elem.align > maxAlign {
  10.         throw("makechan: bad alignment")
  11.     }
  12.     // 计算需要分配的内存大小，如果申请的内存过大，就panic
  13.     mem, overflow := math.MulUintptr(elem.size, uintptr(size))
  14.     if overflow || mem > maxAlloc-hchanSize || size < 0 {
  15.         panic(plainError("makechan: size out of range"))
  16.     }
  18.     var c *hchan
  19.     switch {
  20.     case mem == 0:
  21.         // 申请的是0缓冲队列
  22.         c = (*hchan)(mallocgc(hchanSize, nil, true))
  23.         // Race detector uses this location for synchronization.
  24.         c.buf = c.raceaddr()
  25.     case elem.kind&kindNoPointers != 0:
  26.         // 元素不包含指针
  27.         // Allocate hchan and buf in one call.
  28.         c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
  29.         c.buf = add(unsafe.Pointer(c), hchanSize)
  30.     default:
  31.         // 元素包含指针
  32.         c = new(hchan)
  33.         c.buf = mallocgc(mem, elem, true)
  34.     }
  36.     c.elemsize = uint16(elem.size)
  37.     c.elemtype = elem
  38.     c.dataqsiz = uint(size)
  40.     ...
  42.     return c
  43. }

根据makechan的源码，我们了解到创建一个channel的本质是创建一个hchan结构体，我们通过makechan拿到的就是这个hchan结构体的指针。makechan具体做了以下几个事情：

  1. 前置检查，检查元素大小，以及申请分配的内存空间是否合理。
  2. 根据channel类型触发相应的内存分配，无缓冲长度的channel不分配内存给环形缓冲区；带缓冲区长度的channel调用不同的方法创建缓冲区，分配相应内存。
  3. hchan结构体赋值，返回hchan指针。

channel的数据发送

当执行c <- 1这样的代码时，即向channel发送数据，实际调用的函数是chansend。这里的chansend做了以下几件事情：

  1. 先加锁，保证并发读写安全
  2. 检查channel是否关闭，如果向已关闭的channel发送数据，触发panic
  3. 检查recvq是否有等待数据的goroutine，如果有，则取出recvq头部的goroutine，调用send，把发送方的数据直接拷贝给接收方，而无需拷贝到环形缓冲区。
  4. 如果环形缓冲队列未满，则把要发送的数据拷贝到环形缓冲区，更新环形队列的sendx值。
  5. 环形队列已满时，如果是非阻塞模式，则立即返回；如果是阻塞模式，那就把这个goroutine先挂起放到sendq等待数据。
  1. func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  2.     if c == nil {
  3.         if !block {
  4.             return false
  5.         }
  6.         gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
  7.         throw("unreachable")
  8.     }
  10.     if debugChan {
  11.         print("chansend: chan=", c, "\n")
  12.     }
  14.     if raceenabled {
  15.         racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
  16.     }
  18.     // ready for sending and then observe that it is not closed, that implies that the
  19.     // channel wasn't closed during the first observation.
  20.     // 这里的判断没有加锁，而是使用了
  21.     if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
  22.         (c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {
  23.         return false
  24.     }
  26.     var t0 int64
  27.     if blockprofilerate > 0 {
  28.         t0 = cputicks()
  29.     }
  31.     // chansend操作全程加锁
  32.     lock(&c.lock)
  34.     // channel已关闭，继续往里send会panic
  35.     if c.closed != 0 {
  36.         unlock(&c.lock)
  37.         panic(plainError("send on closed channel"))
  38.     }
  40.     // 检查recvq是否有等待数据的goroutine，如果有，取出recvq头部的goroutine，调用send，把发送方的数据直接拷贝给接收方，而无需拷贝到环形缓冲区
  41.     if sg := c.recvq.dequeue(); sg != nil {
  42.         // Found a waiting receiver. We pass the value we want to send
  43.         // directly to the receiver, bypassing the channel buffer (if any).
  44.         send(c, sg, ep, func() { unlock(&c.lock) }, 3)
  45.         return true
  46.     }
  48.     // 环形队列未满，则将要发送的数据放入环形队列，更新sendx值。
  49.     if c.qcount < c.dataqsiz {
  50.         // Space is available in the channel buffer. Enqueue the element to send.
  51.         qp := chanbuf(c, c.sendx)
  52.         if raceenabled {
  53.             raceacquire(qp)
  54.             racerelease(qp)
  55.         }
  56.         typedmemmove(c.elemtype, qp, ep)
  57.         c.sendx++
  58.         if c.sendx == c.dataqsiz {
  59.             c.sendx = 0
  60.         }
  61.         c.qcount++
  62.         unlock(&c.lock)
  63.         return true
  64.     }
  66.     // 下面就是环形队列已满的情况
  69.     // 如果是非阻塞模式，此时就可以返回了
  70.     if !block {
  71.         unlock(&c.lock)
  72.         return false
  73.     }
  75.     // 如果是阻塞模式，因为队列已满，但是此时又需要发送数据到该队列，那就把这个goroutine先挂起放到sendq
  77.     gp := getg()
  78.     mysg := acquireSudog()
  79.     mysg.releasetime = 0
  80.     if t0 != 0 {
  81.         mysg.releasetime = -1
  82.     }
  83.     // No stack splits between assigning elem and enqueuing mysg
  84.     // on gp.waiting where copystack can find it.
  85.     mysg.elem = ep
  86.     mysg.waitlink = nil
  87.     mysg.g = gp
  88.     mysg.isSelect = false
  89.     mysg.c = c
  90.     gp.waiting = mysg
  91.     gp.param = nil
  92.     c.sendq.enqueue(mysg) // 将当前要发送数据的goroutine挂起放到sendq
  93.     goparkunlock(&c.lock, waitReasonChanSend, traceEvGoBlockSend, 3)
  95.     KeepAlive(ep)
  97.     // someone woke us up.
  98.     if mysg != gp.waiting {
  99.         throw("G waiting list is corrupted")
  100.     }
  101.     gp.waiting = nil
  102.     if gp.param == nil {
  103.         if c.closed == 0 {
  104.             throw("chansend: spurious wakeup")
  105.         }
  106.         panic(plainError("send on closed channel"))
  107.     }
  108.     gp.param = nil
  109.     if mysg.releasetime > 0 {
  110.         blockevent(mysg.releasetime-t0, 2)
  111.     }
  112.     mysg.c = nil
  113.     releaseSudog(mysg)
  114.     return true
  115. }

channel的数据接收

当执行a <- c这样的代码时，即从channel读取数据时，实际调用的函数是chanrecv。这里的chanrecv做了以下几件事情：

  1. 首先加锁
  2. 如果环形缓冲区已满，且sendq有goroutine在的等待发送数据，直接调用recv去取sendq头部goroutine，将其要发送的数据拷贝给自己，不走环形缓冲区；
  3. 环形缓冲区有数据但队列未满，就消费环形队列里的头部元素；
  4. 环形缓冲区没有数据，非阻塞模式下就直接返回；阻塞模式下，此时需要把goroutine放入recvq等待数据。
  1. func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
  2.     // raceenabled: don't need to check ep, as it is always on the stack
  3.     // or is new memory allocated by reflect.
  5.     if debugChan {
  6.         print("chanrecv: chan=", c, "\n")
  7.     }
  9.     if c == nil {
  10.         if !block {
  11.             return
  12.         }
  14.         gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
  15.         throw("unreachable")
  16.     }
  18.     // The order of operations is important here: reversing the operations can lead to
  19.     // incorrect behavior when racing with a close.
  20.     if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||
  21.         c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&
  22.         atomic.Load(&c.closed) == 0 {
  23.         return
  24.     }
  26.     var t0 int64
  27.     if blockprofilerate > 0 {
  28.         t0 = cputicks()
  29.     }
  31.     lock(&c.lock)
  33.     if c.closed != 0 && c.qcount == 0 {
  34.         if raceenabled {
  35.             raceacquire(c.raceaddr())
  36.         }
  37.         unlock(&c.lock)
  38.         if ep != nil {
  39.             typedmemclr(c.elemtype, ep)
  40.         }
  41.         return true, false
  42.     }
  44.     // 值得注意，如果channel已经关闭了，我们还是可以走下面的逻辑，即数据读取
  46.     // 此时环形缓冲区已满，sendq有goroutine在的等待发送数据，直接调用recv去取sendq头部goroutine，将其要发送的数据拷贝给自己，不走环形缓冲区
  47.     if sg := c.sendq.dequeue(); sg != nil {
  48.         recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
  49.         return true, true
  50.     }
  52.     // 环形缓冲区有数据但队列未满，就消费环形队列里的头部元素
  53.     if c.qcount > 0 {
  54.         // Receive directly from queue
  55.         qp := chanbuf(c, c.recvx)
  56.         if raceenabled {
  57.             raceacquire(qp)
  58.             racerelease(qp)
  59.         }
  60.         if ep != nil {
  61.             typedmemmove(c.elemtype, ep, qp) //数据拷贝，sender->reciever
  62.         }
  63.         typedmemclr(c.elemtype, qp) // 清理sender的发送数据内存
  64.         c.recvx++
  65.         if c.recvx == c.dataqsiz {
  66.             c.recvx = 0
  67.         }
  68.         c.qcount--
  69.         unlock(&c.lock)
  70.         return true, true
  71.     }
  73.     if !block {
  74.         unlock(&c.lock)
  75.         return false, false
  76.     }
  78.     // 环形缓冲区没有数据，此时需要把goroutine放入recvq等待数据
  79.     // no sender available: block on this channel.
  80.     gp := getg()
  81.     mysg := acquireSudog()
  82.     mysg.releasetime = 0
  83.     if t0 != 0 {
  84.         mysg.releasetime = -1
  85.     }
  86.     // No stack splits between assigning elem and enqueuing mysg
  87.     // on gp.waiting where copystack can find it.
  88.     mysg.elem = ep
  89.     mysg.waitlink = nil
  90.     gp.waiting = mysg
  91.     mysg.g = gp
  92.     mysg.isSelect = false
  93.     mysg.c = c
  94.     gp.param = nil
  95.     c.recvq.enqueue(mysg)
  96.     goparkunlock(&c.lock, waitReasonChanReceive, traceEvGoBlockRecv, 3)
  98.     // someone woke us up
  99.     if mysg != gp.waiting {
  100.         throw("G waiting list is corrupted")
  101.     }
  102.     gp.waiting = nil
  103.     if mysg.releasetime > 0 {
  104.         blockevent(mysg.releasetime-t0, 2)
  105.     }
  106.     closed := gp.param == nil
  107.     gp.param = nil
  108.     mysg.c = nil
  109.     releaseSudog(mysg)
  110.     return true, !closed
  111. }

channel的关闭

close(c)用于关闭channel。channel的关闭实际调用的是closechan，注意以下几点：

  1. 关闭已关闭的channel，触发panic
  2. channel关闭时会通知该recvq队列中的所有Reader当前channel已关闭，这些recvq中的g会被释放。
  3. channel关闭时会通知该sendq队列中的所有writer当前channel已关闭，这会触发panic。这些sendq中的g会被释放。
  4. channel关闭操作也是全程加锁。
  1. func closechan(c *hchan) {
  2.     if c == nil {
  3.         panic(plainError("close of nil channel"))
  4.     }
  6.     lock(&c.lock)
  7.     // 关闭已关闭的channel，触发panic
  8.     if c.closed != 0 {
  9.         unlock(&c.lock)
  10.         panic(plainError("close of closed channel"))
  11.     }
  13.     if raceenabled {
  14.         callerpc := getcallerpc()
  15.         racewritepc(c.raceaddr(), callerpc, funcPC(closechan))
  16.         racerelease(c.raceaddr())
  17.     }
  19.     c.closed = 1
  21.     var glist gList
  23.     // 通知所有该channel的Reader这个channel关闭了，释放recvq中的所有g
  24.     // release all readers
  25.     for {
  26.         sg := c.recvq.dequeue()
  27.         if sg == nil {
  28.             break
  29.         }
  30.         if sg.elem != nil {
  31.             typedmemclr(c.elemtype, sg.elem)
  32.             sg.elem = nil
  33.         }
  34.         if sg.releasetime != 0 {
  35.             sg.releasetime = cputicks()
  36.         }
  37.         gp := sg.g
  38.         gp.param = nil
  39.         if raceenabled {
  40.             raceacquireg(gp, c.raceaddr())
  41.         }
  42.         glist.push(gp)
  43.     }
  45.     // release all writers (they will panic)
  46.     // 释放sendq中的所有g
  47.     for {
  48.         sg := c.sendq.dequeue()
  49.         if sg == nil {
  50.             break
  51.         }
  52.         sg.elem = nil
  53.         if sg.releasetime != 0 {
  54.             sg.releasetime = cputicks()
  55.         }
  56.         gp := sg.g
  57.         gp.param = nil
  58.         if raceenabled {
  59.             raceacquireg(gp, c.raceaddr())
  60.         }
  61.         glist.push(gp)
  62.     }
  63.     unlock(&c.lock)
  65.     // 释放这些挂起的g
  66.     for !glist.empty() {
  67.         gp := glist.pop()
  68.         gp.schedlink = 0
  69.         goready(gp, 3) 
  70.     }
  71. }

如果channel已经关闭，但此时还有goroutine去接收该channel数据，如果channel的环形缓冲队列里已无数据，则x,ok := <- ch中，x为nil，ok为false；如果channel的环形缓冲队列还有数据，此时去接收数据时可以收到数据的，x为接收到的数据，ok为true。只有当channel关闭且无数据时，x才为nil，ok为false。