作用
- Go 语言的 channel 是一种 goroutine 之间的通信方式，它可以用来传递数据，也可以用来同步 goroutine 的执行。
- chan 是 goroutine 之间的通信桥梁，可以安全地在多个 goroutine 中共享数据。
- 使用 chan 实现 goroutine 之间的协作与同步，可用于信号传递、任务完成通知等。
- select 配合 chan，可以同时监听多个 channel，处理任意一个可用 channel 的数据。
结构
type hchan struct { qcount uint // 队列中的元素个数 dataqsiz uint // 环形队列的容量 buf unsafe.Pointer // 环形队列的指针 elemsize uint16 // 元素的大小 closed uint32 // 是否关闭 如果以关闭则不是0 timer *timer // 为此 channel 提供时间控制的计时器 elemtype *_type // 元素的类型 sendx uint // 发送索引，指示下一个发送操作的位置 recvx uint // 接收索引，指示下一个接收操作的位置 recvq waitq // 等待接收的等待队列 sendq waitq // 等待发送的等待队列 // 锁 lock mutex } 
waitq
type waitq struct { first *sudog // 首指针 last *sudog // 尾指针 } sudog
type sudog struct { g *g // goroutine next *sudog // 指向下一个sudog，用于形成链表 prev *sudog // 指向上一个sudog，用于形成链表 elem unsafe.Pointer // 指向数据元素的指针（可能指向栈上的数据） acquiretime int64 // 获取资源的时间 releasetime int64 // 释放资源的时间 ticket uint32 // 票据号码，用于排序和公平性 isSelect bool // 标志是否在select操作中使用此sudog success bool // 通信是否成功（接收到值或因 channel 关闭被唤醒） waiters uint16 // 等待者数量，仅在列表头部有意义 parent *sudog // 指向父节点的指针，在二叉树结构中使用 waitlink *sudog // g的等待链表或semaRoot waittail *sudog // semaRoot的尾部 c *hchan // 指向sudog所等待的 channel } 创建
创建一个 channel:
func makechan(t *chantype, size int) *hchan { // 元素类型 elem := t.Elem // 检查大小是否合法 if elem.Size_ >= 1<<16 { throw("makechan: invalid channel element type") } // 是否满足对齐要求 if hchanSize%maxAlign != 0 || elem.Align_ > maxAlign { throw("makechan: bad alignment") } // 计算内存分配所需大小：`元素大小 * 数量`。 mem, overflow := math.MulUintptr(elem.Size_, uintptr(size)) if overflow || mem > maxAlloc-hchanSize || size < 0 { panic(plainError("makechan: size out of range")) } var c *hchan switch { case mem == 0: // 队列大小为0 说明是无缓冲的channel 直接分配hchan // 分配内存 hchanSize 是 hchan 结构体的大小 c = (*hchan)(mallocgc(hchanSize, nil, true)) c.buf = c.raceaddr() case !elem.Pointers(): // 如果元素中不包含指针 则使用一个连续的内存块 结构体和 buf 是连续的 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) lockInit(&c.lock, lockRankHchan) if debugChan { print("makechan: chan=", c, "; elemsize=", elem.Size_, "; dataqsiz=", size, "n") } return c } const hchanSize = unsafe.Sizeof(hchan{}) + uintptr(-int(unsafe.Sizeof(hchan{}))&(maxAlign-1)) func (c *hchan) raceaddr() unsafe.Pointer { // 将对 channel 的读写操作视为发生在这个地址。 // 避免使用 `qcount` 或 `dataqsiz` 的地址， // 因为内建函数 `len()` 和 `cap()` 会读取这些地址， // 而我们不希望这些操作与例如 `close()` 之类的操作发生竞争。 return unsafe.Pointer(&c.buf) } 写
func chansend1(c *hchan, elem unsafe.Pointer) { chansend(c, elem, true, getcallerpc()) } func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool { if c == nil { if !block { return false } // 如果 channel 为空 挂起当前 goroutine 并报错 gopark(nil, nil, waitReasonChanSendNilChan, traceBlockForever, 2) throw("unreachable") } // ...... // 检查非阻塞模式是否可以直接返回失败结果 if !block && c.closed == 0 && full(c) { return false } var t0 int64 if blockprofilerate > 0 { t0 = cputicks() } lock(&c.lock) // 检查 channel 是否已经关闭 if c.closed != 0 { unlock(&c.lock) panic(plainError("send on closed channel")) } if sg := c.recvq.dequeue(); sg != nil { // 如果有等待接收的 Goroutine，直接将值发送给它，跳过缓冲区 send(c, sg, ep, func() { unlock(&c.lock) }, 3) return true } if c.qcount < c.dataqsiz { // 如果通道缓冲区有空间，直接将值写入缓冲区 qp := chanbuf(c, c.sendx) if raceenabled { racenotify(c, c.sendx, nil) } typedmemmove(c.elemtype, qp, ep) c.sendx++ if c.sendx == c.dataqsiz { c.sendx = 0 } c.qcount++ unlock(&c.lock) return true } if !block { // 非阻塞模式且无法发送值，返回 false unlock(&c.lock) return false } // 阻塞模式，当前 Goroutine 挂起等待接收者 gp := getg() // 放入 acquireSudog 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) gp.parkingOnChan.Store(true) gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceBlockChanSend, 2) // 确保发送值在接收者拷贝之前不会被释放 KeepAlive(ep) // 唤醒后，检查状态 if mysg != gp.waiting { throw("G waiting list is corrupted") } gp.waiting = nil gp.activeStackChans = false closed := !mysg.success gp.param = nil if mysg.releasetime > 0 { blockevent(mysg.releasetime-t0, 2) } mysg.c = nil // 回收 sudog releaseSudog(mysg) if closed { if c.closed == 0 { throw("chansend: spurious wakeup") } panic(plainError("send on closed channel")) } return true } 所以阻塞写这个主要有三种模式:
- 如果有等待接收的 Goroutine （c.recvq 里面有值），说明 buf 要么满了 要么就没有，直接将值发送给它，跳过缓冲区
- 如果通道缓冲区有空间，直接将值写入缓冲区
- 如果缓冲区没有空间，且是阻塞模式，当前 Goroutine 挂起等待接收者
send
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) { // ...... // 如果接收者有一个有效的元素指针，则将发送者的数据直接拷贝给接收者 if sg.elem != nil { sendDirect(c.elemtype, sg, ep) sg.elem = nil } gp := sg.g unlockf() gp.param = unsafe.Pointer(sg) sg.success = true if sg.releasetime != 0 { sg.releasetime = cputicks() } // 唤醒接收者 Goroutine goready(gp, skip+1) } func sendDirect(t *_type, sg *sudog, src unsafe.Pointer) { // 内存拷贝 dst := sg.elem typeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.Size_) memmove(dst, src, t.Size_) } 大致的逻辑为，取出 goroutine 然后把接受的值 COPY 到接受者的内存中，然后唤醒接受者 goroutine。
接受的内存可能是堆也可能是栈，堆还好说，如果是栈，就是在一个栈内直接操作其他的栈了，按理来说，这是不安全的。但是，这是 runtime, 我们已经把 goroutine GoPark 了，保证了它不会执行，所以这里是安全的。当然我们自己写代码时，肯定是不能这么做的。
chanbuf && typedmemmove
func chanbuf(c *hchan, i uint) unsafe.Pointer { // 在 buf 上加上 i * elemsize 的偏移量 return add(c.buf, uintptr(i)*uintptr(c.elemsize)) } func typedmemmove(typ *abi.Type, dst, src unsafe.Pointer) { if dst == src { return } if writeBarrier.enabled && typ.Pointers() { // 如果写屏障启用且类型包含指针，则需要处理写屏障。 bulkBarrierPreWrite(uintptr(dst), uintptr(src), typ.PtrBytes, typ) } // 执行内存拷贝 memmove(dst, src, typ.Size_) if goexperiment.CgoCheck2 { cgoCheckMemmove2(typ, dst, src, 0, typ.Size_) } } acquireSudog & releaseSudog
func acquireSudog() *sudog { mp := acquirem() pp := mp.p.ptr() // 如果 sudog 缓存为空，需要补充缓存 if len(pp.sudogcache) == 0 { lock(&sched.sudoglock) for len(pp.sudogcache) < cap(pp.sudogcache)/2 && sched.sudogcache != nil { s := sched.sudogcache sched.sudogcache = s.next s.next = nil pp.sudogcache = append(pp.sudogcache, s) } unlock(&sched.sudoglock) if len(pp.sudogcache) == 0 { pp.sudogcache = append(pp.sudogcache, new(sudog)) } } // 从 P 的缓存中取出一个 sudog n := len(pp.sudogcache) s := pp.sudogcache[n-1] pp.sudogcache[n-1] = nil pp.sudogcache = pp.sudogcache[:n-1] if s.elem != nil { throw("acquireSudog: found s.elem != nil in cache") } releasem(mp) return s } func releaseSudog(s *sudog) { // ...... gp := getg() mp := acquirem() pp := mp.p.ptr() if len(pp.sudogcache) == cap(pp.sudogcache) { // 如果本地缓存已满，将部分 sudog 转移到全局缓存 var first, last *sudog for len(pp.sudogcache) > cap(pp.sudogcache)/2 { n := len(pp.sudogcache) p := pp.sudogcache[n-1] pp.sudogcache[n-1] = nil pp.sudogcache = pp.sudogcache[:n-1] if first == nil { first = p } else { last.next = p } last = p } lock(&sched.sudoglock) last.next = sched.sudogcache sched.sudogcache = first unlock(&sched.sudoglock) } // 将 sudog 放回本地缓存 pp.sudogcache = append(pp.sudogcache, s) releasem(mp) } 读
func chanrecv1(c *hchan, elem unsafe.Pointer) { chanrecv(c, elem, true) } // 带 ok 时 比如 `v, ok := <-ch` func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) { _, received = chanrecv(c, elem, true) return } func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) { // ...... // 非阻塞模式下检查失败条件 if !block && empty(c) { if atomic.Load(&c.closed) == 0 { // 已经没关闭 直接返回 因为这是非阻塞模式而且 buf 为空的情况 return } if empty(c) { if raceenabled { raceacquire(c.raceaddr()) } if ep != nil { typedmemclr(c.elemtype, ep) } return true, false } } var t0 int64 if blockprofilerate > 0 { t0 = cputicks() } lock(&c.lock) if c.closed != 0 { // 通道已关闭 检查是否有数据 if c.qcount == 0 { if raceenabled { raceacquire(c.raceaddr()) } unlock(&c.lock) if ep != nil { // 把数据清零 因为通道已经关闭了 typedmemclr(c.elemtype, ep) } return true, false } } else { // 通道未关闭，检查是否有等待发送的 Goroutine if sg := c.sendq.dequeue(); sg != nil { // 直接从发送队列中取出值 recv(c, sg, ep, func() { unlock(&c.lock) }, 3) return true, true } } // 如果缓冲区中有数据，从缓冲区接收 if c.qcount > 0 { qp := chanbuf(c, c.recvx) if raceenabled { racenotify(c, c.recvx, nil) } if ep != nil { // 直接 COPY 内存 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 } // 如果是非阻塞接收，直接返回 if !block { unlock(&c.lock) return false, false } // 没有可用的发送方：阻塞在该通道上 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) if c.timer != nil { blockTimerChan(c) } gp.parkingOnChan.Store(true) gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceBlockChanRecv, 2) // someone woke us up if mysg != gp.waiting { throw("G waiting list is corrupted") } if c.timer != nil { unblockTimerChan(c) } gp.waiting = nil gp.activeStackChans = false if mysg.releasetime > 0 { blockevent(mysg.releasetime-t0, 2) } success := mysg.success gp.param = nil mysg.c = nil releaseSudog(mysg) return true, success } 所以读数据（阻塞读）的逻辑为:
- 检查 channel 是否已经关闭 如果关闭了 而且没有数据了 直接返回
- 如果有等待发送的 Goroutine （c.sendq 里面有值），如果无缓冲chan 直接从goroutine中取值 负责从 buf 取出值 并把数据加入末尾
- 如果缓冲区中有数据，从缓冲区接收
- 如果缓冲区没有数据了 挂起 goroutine 并加入 recvq 等待接收者
recv
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) { if c.dataqsiz == 0 { if ep != nil { // 如果是无缓冲通道，直接 Copy 数据 recvDirect(c.elemtype, sg, ep) } } else { // 否则，通道是有缓冲通道。 // 从队列的头部获取数据，同时通知发送方将其数据放到尾部 qp := chanbuf(c, c.recvx) // 从队列复制数据到接收方 if ep != nil { typedmemmove(c.elemtype, ep, qp) } // 将发送者的数据复制到队列中 typedmemmove(c.elemtype, qp, sg.elem) c.recvx++ if c.recvx == c.dataqsiz { c.recvx = 0 } c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz } sg.elem = nil gp := sg.g unlockf() gp.param = unsafe.Pointer(sg) sg.success = true if sg.releasetime != 0 { sg.releasetime = cputicks() } goready(gp, skip+1) } 关闭
func closechan(c *hchan) { // 空值检查 if c == nil { panic(plainError("close of nil channel")) } lock(&c.lock) // 如果已经关闭了 报错 不能关闭已经关闭的 channel if c.closed != 0 { unlock(&c.lock) panic(plainError("close of closed channel")) } if raceenabled { callerpc := getcallerpc() racewritepc(c.raceaddr(), callerpc, abi.FuncPCABIInternal(closechan)) racerelease(c.raceaddr()) } // 设置状态 c.closed = 1 // 创建一个 G 列表，用于保存需要唤醒的 Goroutine 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 = unsafe.Pointer(sg) sg.success = false if raceenabled { raceacquireg(gp, c.raceaddr()) } glist.push(gp) } // 释放所有的写方 会 panic 因为向已经关闭的 channel 写数据是不允许的 for { sg := c.sendq.dequeue() if sg == nil { break } sg.elem = nil if sg.releasetime != 0 { sg.releasetime = cputicks() } gp := sg.g gp.param = unsafe.Pointer(sg) sg.success = false if raceenabled { raceacquireg(gp, c.raceaddr()) } glist.push(gp) } unlock(&c.lock) // 唤醒所有的 Goroutine for !glist.empty() { gp := glist.pop() gp.schedlink = 0 goready(gp, 3) } } 非阻塞读写
非阻塞的方式一般用在 select 中。
func selectnbsend(c *hchan, elem unsafe.Pointer) (selected bool) { return chansend(c, elem, false, getcallerpc()) } func selectnbrecv(elem unsafe.Pointer, c *hchan) (selected, received bool) { return chanrecv(c, elem, false) } - 在阻塞下，需要当前 goroutine 挂起时，非阻塞则不需要，直接返回 flase。
- 如果能直接读数据，则返回 true。
select
func walkSelectCases(cases []*ir.CommClause) []ir.Node { // ...... switch n.Op() { default: base.Fatalf("select %v", n.Op()) case ir.OSEND: // if selectnbsend(c, v) { body } else { default body } n := n.(*ir.SendStmt) ch := n.Chan cond = mkcall1(chanfn("selectnbsend", 2, ch.Type()), types.Types[types.TBOOL], r.PtrInit(), ch, n.Value) case ir.OSELRECV2: n := n.(*ir.AssignListStmt) recv := n.Rhs[0].(*ir.UnaryExpr) ch := recv.X elem := n.Lhs[0] if ir.IsBlank(elem) { elem = typecheck.NodNil() } cond = typecheck.TempAt(base.Pos, ir.CurFunc, types.Types[types.TBOOL]) fn := chanfn("selectnbrecv", 2, ch.Type()) call := mkcall1(fn, fn.Type().ResultsTuple(), r.PtrInit(), elem, ch) as := ir.NewAssignListStmt(r.Pos(), ir.OAS2, []ir.Node{cond, n.Lhs[1]}, []ir.Node{call}) r.PtrInit().Append(typecheck.Stmt(as)) } // ...... } func selectnbsend(c *hchan, elem unsafe.Pointer) (selected bool) { return chansend(c, elem, false, getcallerpc()) } func selectnbrecv(elem unsafe.Pointer, c *hchan) (selected, received bool) { return chanrecv(c, elem, false) } 改写后就是调用 selectnbsend 非阻塞的从 channel 发送数据，如果成功则返回 true，否则返回 false。失败了就从下个 case 继续执行。
