Go 源码之 chan

go源码之chan - Jxy 博客

一、总结

chan 提供了一种在 goroutine 之间进行数据交换和同步的方式。通道可以用于控制并发访问和共享数据，从而减少竞态条件和死锁问题，并且可以自然地处理异步事件和信号。如果你的应用程序需要在 goroutine 之间传递数据或消息，那么通道是一个不错的选择

• 内部是一个 hchan 结构（字段见源码），环形队列 + 发送者双向链表 + 接收者双向链表 + 锁
• channel 与 select 语句结合使用时，底层调用的还是 chansend 和 chanrecv 函数
• channel
  • 结构：环形缓存、sendq、recvq；
  • 流程：上锁/解锁，阻塞/非阻塞，缓冲/非缓冲，缓存入队出队，sudog 入队出队，协程休眠/唤醒

二、源码

/src/runtime/chan.go

• 一个环形队列
• 两个双向列表

（一）hchan

buf + sendx + recvx 形成环形队列

type hchan struct {qcount   uint           // 队列中现存元素数量dataqsiz uint           // 队列容量（缓冲区）buf      unsafe.Pointer // 队列，指向一个动态分配的数组，用于存储 channel 中的元素elemsize uint16         // 队列中元素大小closed   uint32         // 0 正常	，1 关闭elemtype *_type     		// 队列中元素类型，sendx    uint   				// 队列（buf）已发送位置，当（sendx++）==dataqsiz，则从头开始发，sendx=0recvx    uint   				// 队列（buf）已接收位置；// 当 `sendx` 和 `recvx` 相等时，channel 中无元素，发送 / 接收 操作阻塞recvq    waitq  				// 双向链表 ，FIFO 由 recv 行为（也就是读 <-ch）阻塞在 channel 上的 goroutine 队列sendq    waitq 					// 双向链表 ，FIFO 由 send 行为 (也就是写 ch<-) 阻塞在 channel 上的 goroutine 队列lock mutex 							// 读写锁，保护hchan中的所有字段，以及waitq中所有的字段
}// 双向链表，存储了g
type waitq struct {first *sudog  // 链表头部，协程 g 的数据结构last  *sudog  // 链表尾部，协程 g 的数据结构
}

（二）创建

ch1 := make(chan int)
ch2 := make(chan int,2)

底层都是调用了runtime.makechan()

• 合法性校验
  • 数据类型大小校验
  • 内存溢出校验
• 初始化 hchan
  • 初始化 无缓冲 hchan
  • 初始化 有缓冲 && 无指针元素 hchan
  • 初始化 无缓冲 && 有指针元素 hchan
  • 初始化 hchan 其他元素：如 dataqsize、elemsize、elemtype、lock

// 主要逻辑：合法性验证 和 分配地址空间
// t 是指向 chantype 的指针，size 表示缓冲区大小，0表无缓冲
func makechan(t *chantype, size int) *hchan {elem := t.elem // 元素的类型// ----------- 1. 合法性验证 ----------// 数据类型大小验证，大于1<<16时异常if elem.size >= 1<<16 {throw("makechan: invalid channel element type")}// 内存对齐（降低寻址次数），大于最大内存（8字节数）时异常if hchanSize%maxAlign != 0 || elem.align > maxAlign {throw("makechan: bad alignment")}// 传入的size大于堆可分配的最大内存时：内存溢出异常mem, overflow := math.MulUintptr(elem.size, uintptr(size))if overflow || mem > maxAlloc-hchanSize || size < 0 {panic(plainError("makechan: size out of range"))}// ----------- 2. 分配地址空间 ----------// hchanSize 为 hchan 结构大小// mem 为缓存区大小/* 根据 channel 中收发元素的类型和缓冲区的大小初始化 runtime.hchan 和 缓冲区，分为三种情况：* 如果不存在缓冲区，分配 hchan 结构体空间，即无缓存 channel* 如果 channel 存储的类型不是指针类型，分配连续地址空间，包括 hchan 结构体 + 数据* 默认情况包括指针，为 hchan 和 buf 单独分配数据地址空间更新 hchan 结构体的数据，包括 elemsize、elemtype 和 dataqsiz*/var c *hchanswitch {case mem == 0:// 创建无缓冲的 chan ，buf==0 ，初始化 hchanc = (*hchan)(mallocgc(hchanSize, nil, true)) // hchanSize表示空hchan需要占用的字节c.buf = c.raceaddr()  //  raceaddr内部实现为：return unsafe.Pointer(&c.buf)case elem.ptrdata == 0:// 有缓存区，并且队列中不存在指针，分配连续地址空间，大小为 hchanSize + memc = (*hchan)(mallocgc(hchanSize+mem, nil, true))// buf指针指向空hchan占用空间的末尾c.buf = add(unsafe.Pointer(c), hchanSize)default:// 队列包含指针类型// 为buf单独开辟mem大小的空间，用来保存所有的数据c = new(hchan)c.buf = mallocgc(mem, elem, true)}c.elemsize = uint16(elem.size)  			// 元素大小c.elemtype = elem 										// 元素类型c.dataqsiz = uint(size) 							// chan 缓存区大小lockInit(&c.lock, lockRankHchan) 			// 初始化锁if debugChan {print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")}return c
}

（三）发送

ch <- 1

执行 runtime.chansend1(SB)

• 异常检查：
  • 发送到 nil chan 中，阻塞挂起
  • 往 closed chan 发送（写），则 panic
  • 当前 chan 是否可以发送
• 同步发送：recvq 中存在等待接收者，则直接唤醒并发送数据
• 异步发送：c.qcount < c.dataqsiz 缓存区空闲，则数据发送到缓存区
• 阻塞发送：当前面都不满足时 且 block = true 时：发送操作 线程阻塞 挂起，并且添加到 sendq 等待队列，直到有接收者接收才释放

/**
* @Description: chansend函数主要可以归纳为四部分：异常检查、同步发送、异步发送、阻塞发送：
* @Param:c：hchan结构；ep：发送的元素；block：是否阻塞；callerpc：
* @return: true发送成功，false发送失败
*/
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {// ------------------------------------ 1.异常检查 ------------------------------------if c == nil { if !block {return false}gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2) 	// 发送到 nil chan 中，阻塞挂起throw("unreachable")}..........// 当channel不为nil，此时检查channel是否做好接收发送操作的准备，if !block && c.closed == 0 && full(c) { 	return false															// 非阻塞且未关闭: 1. 无缓存区，recvq为空 2. 有缓冲区，但是buffer已满}lock(&c.lock) // 先上锁if c.closed != 0 { // chan已经关闭，则解锁unlock(&c.lock)panic(plainError("send on closed channel")) 				// 往 closed chan 发送（写），则 panic}// ------------------------------------  2.同步发送 ------------------------------------// recvq 中存在等待接收者，则直接唤醒并发送数据if sg := c.recvq.dequeue(); sg != nil { send(c, sg, ep, func() { unlock(&c.lock) }, 3)		// recvq 等待队列取取出 sg（sudog）并唤醒并发送数据 epreturn true}// ------------------------------------  3.异步发送 ------------------------------------// （有缓存区，没有等待接收者，先发到缓冲区中，等有接收者再去读）if c.qcount < c.dataqsiz {  					// 存在的元素个数< 缓冲区：说明缓存区可以继续写数据qp := chanbuf(c, c.sendx) 					// 获取缓存区index地址typedmemmove(c.elemtype, qp, ep)		// 数据写入bufferc.sendx++ 													// 发送数据的下标++if c.sendx == c.dataqsiz { 					// 当发送数据的下标等于缓冲区，表数据发送完毕，从头开始c.sendx = 0}c.qcount++ 													// 元素数量++unlock(&c.lock) 										// 解锁return true 												// 返回结果}if !block {unlock(&c.lock) 										// 解锁return false}// ------------------------------------ 4. 阻塞发送 ------------------------------------// 当前面都不满足时（没有等待接收者，没有空闲缓冲区） 且 block = true 时，发送操作 线程阻塞 挂起，直到有接收者接收才释放：gp := getg()..........c.sendq.enqueue(mysg)  											// 将发送 的 sg 添加到 sendq 等待队列中return true
}

func full(c *hchan) bool {if c.dataqsiz == 0 { // 无缓冲return c.recvq.first == nil}// 有缓冲，现有元素的个数 是否等于 缓冲区容量时（缓冲区满）return c.qcount == c.dataqsiz
}

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.gunlockf()gp.param = unsafe.Pointer(sg)sg.success = trueif sg.releasetime != 0 {sg.releasetime = cputicks()}goready(gp, skip+1) // 唤醒sudog协程；下一轮调度时会唤醒这个接收的 goroutine。
}

// 实现了等待队列的入队操作。它将一个元素添加到等待队列的末尾，并更新队列的 first 和 last 指针
func (q *waitq) enqueue(sgp *sudog) {sgp.next = nil // 表示该元素是队列的最后一个x := q.last // 将等待队列 q 中的最后一个元素（如果存在）赋值给变量 x。if x == nil { // 如果队列中最后一个都没有，则队列无元素，即 x 为 nil，sgp.prev = nil // 则将 sgp 元素的 prev 指针设为 nil，表示该元素是队列中的第一个元素，q.first = sgp //  然后将队列的 first 和 last 指针都指向该元素，表示该元素是队列中唯一的元素。q.last = sgp  // 然后直接返回，结束入队操作。return}sgp.prev = x // sgp 是新加的最后元素，需要关联前一个元素（x为原队列中最后一个元素）x.next = sgp // 设置x的的下一个元素为新加的元素q.last = sgp // 设置q队列的最后一个元素
}

（四）接收

i <- ch i, ok <- ch

执行 runtime.chanrecv1(SB) 都是调用的chanrecv()

• 异常检查：
  • 从 nil chan 中读取，阻塞挂起
  • 从 closed chan 接收（读），返回零值
  • 当前 chan 是否可以接收
• 同步接收：sendq 中存在发送者，则直接唤醒并接收数据
• 异步接收：c.qcount 队列中有元素，则则从 buf 中读取数据
• 阻塞接收：当前面都不满足时 且 block = true 时：接收操作 线程阻塞 挂起，并且添加到 recvq 等待队列，直到有发送者才释放

chanrecv 函数的逻辑和 chansend 的逻辑基本一致

（五）关闭

close(ch)

closechan(c *hchan)

主要逻辑：

• 异常检查：
  • 关闭 nil chan ，panic
  • 关闭 closed chan，panic
• 标记 chan 为关闭状态
• 释放等待的 sudog： 唤醒并调度等待队列 recvq、sendq 中的 sudog，所有接收者收到零值

func closechan(c *hchan) {// ------------------------------------ 1. 异常检查 ------------------------------------if c == nil {panic(plainError("close of nil channel")) 				// 关闭 nil chan ，panic}lock(&c.lock)																			// 上锁if c.closed != 0 {unlock(&c.lock)panic(plainError("close of closed channel"))		// 关闭 closed chan，panic}c.closed = 1																			// 标识chan已经关闭// ------------------------------------ 2. 释放等待的 sudog ------------------------------------var glist gList 																	// 存储 recvq、sendq 等待队列中的 sg（sudog）for {sg := c.recvq.dequeue()													// 将 recvq 等待队列中的  sg（sudog） 添加到 glist......glist.push(gp)}for {sg := c.sendq.dequeue()													// 将 sendq 等待队列中的  sg（sudog） 添加到 glist......glist.push(gp)}unlock(&c.lock)																		// 解锁for !glist.empty() {															//依次从 glist 中弹出 sg（sudog）并唤醒执行，所有接收者收到零值gp := glist.pop()gp.schedlink = 0goready(gp, 3)}
}

三、常见问题

1.为什么要使用环形队列

chan的内部使用环形队列来存取元素，每次发/收元素时，会根据sendx/recvx记录的位置从队列buf中存取元素，

所以环形队列：buf+sendx+recvx实现的，

使用环形数组实现的好处：

• 避免对数组进行复制或者移动操作

  比如数组【1，2，3】，现在添加4，变为【2，3，4】，数组就需要进行复制拷贝操作，如果是环形队列，则直接将4添加到队列的尾部即可，

• 避免内存分配和拷贝的开销，从而提高程序的性能

  重复利用，避免重新分配内存

2. 关于chan的操作

有劳各位看官 点赞、关注➕收藏 ，你们的支持是我最大的动力！！！

接下来会不断更新 golang 的一些底层源码及个人开发经验（个人见解）！！！

同时也欢迎大家在评论区提问、分享您的经验和见解！！！