本篇文章基于go1.18版本。 参考了halffrost的博客。https://halfrost.com/

简介

不要通过共享内存进行通信。建议，通过通信来共享内存。（Do not communicate by sharing memory; instead, share memory by communicating）这是 Go 语言并发的哲学座右铭。相比与其他语言，go实现了channel来进行通信，下面我们来看一下channel是如何实现的。

基本结构

channel的底层文件是 src/runtime/chan.go 。

  1. type hchan struct {
  2.     qcount   uint           // 队列中所有数据总数
  3.     dataqsiz uint           // 环形队列的 size
  4.     buf      unsafe.Pointer // 指向 dataqsiz 长度的数组(环形队列)
  5.     elemsize uint16         // 元素大小
  6.     closed   uint32
  7.     elemtype *_type         // 元素类型
  8.     sendx    uint           // 已发送的元素在环形队列中的位置
  9.     recvx    uint           // 已接收的元素在环形队列中的位置
  10.     recvq    waitq          // 接收者的等待队列（双向链表）
  11.     sendq    waitq          // 发送者的等待队列（双向链表）
  13.     lock mutex                //保护资源
  14. }

image.png
图1 channel基本结构

channel 最核心的数据结构是 sudog。sudog 代表了一个在等待队列中的 g。sudog 是 Go 中非常重要的数据结构，因为 g 与同步对象关系是多对多的。一个 g 可以出现在许多等待队列上，因此一个 g 可能有很多sudog。并且多个 g 可能正在等待同一个同步对象，因此一个对象可能有许多 sudog。sudog 是从特殊池中分配出来的。使用 acquireSudog 和 releaseSudog 分配和释放它们。

  1. type sudog struct {
  3.     g *g
  5.     next *sudog
  6.     prev *sudog
  7.     elem unsafe.Pointer // 指向数据 (可能指向栈)
  9.     acquiretime int64
  10.     releasetime int64
  11.     ticket      uint32
  13.     isSelect bool
  14.     success bool
  16.     parent   *sudog     // semaRoot 二叉树
  17.     waitlink *sudog     // g.waiting 列表或者 semaRoot
  18.     waittail *sudog     // semaRoot
  19.     c        *hchan     // channel
  20. }

创建channel

创建 channel 的主要实现在 makechan() 函数中：

  1. func makechan(t *chantype, size int) *hchan {
  2.     elem := t.elem
  4.     // 编译器检查数据项大小不能超过 64KB
  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.     // 缓冲区大小检查，判断是否溢出
  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.         // 队列或者元素大小为 zero 时
  22.         c = (*hchan)(mallocgc(hchanSize, nil, true))
  23.         // Race 竞争检查利用这个地址来进行同步操作
  24.         c.buf = c.raceaddr()
  25.     case elem.ptrdata == 0:
  26.         // 元素不包含指针时。一次分配 hchan 和 buf 的内存。
  27.         c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
  28.         c.buf = add(unsafe.Pointer(c), hchanSize)
  29.     default:
  30.         // 元素包含指针时
  31.         c = new(hchan)
  32.         c.buf = mallocgc(mem, elem, true)
  33.     }
  35.     // 设置属性
  36.     c.elemsize = uint16(elem.size)
  37.     c.elemtype = elem
  38.     c.dataqsiz = uint(size)
  39.     lockInit(&c.lock, lockRankHchan)
  41.     if debugChan {
  42.         print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
  43.     }
  44.     return c
  45. }

发送数据

walkSend() 函数中主要逻辑调用了 chansend1()，而 chansend1() 只是 chansend() 的“外壳”。所以 channel 发送数据的核心实现在 chansend() 中。根据阻塞与否，chansend可以分为四个部分：异常检查、同步发送、异步发送、阻塞发送。

异常检查

  1. func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
  2.     // 判断 channel 是否为 nil
  3.     if c == nil {
  4.         if !block {
  5.             return false
  6.         }
  7.         gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
  8.         throw("unreachable")
  9.     }
  11.     if debugChan {
  12.         print("chansend: chan=", c, "\n")
  13.     }
  15.     if raceenabled {
  16.         racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
  17.     }
  19.     // 简易快速的检查
  20.     if !block && c.closed == 0 && full(c) {
  21.         return false
  22.     }
  23. ......
  24. }

同步发送


异步发送



阻塞发送


接收数据



关闭数据