go语言基础学习九

channel底层实现？是否线程安全#

hchan 结构体#

type hchan struct {
	qcount   uint           // 队列中当前的数据元素总数
	dataqsiz uint           // 循环队列的容量（缓冲区大小）
	buf      unsafe.Pointer // 指向循环队列数组的指针，数组包含dataqsiz个元素
	elemsize uint16         // 通道中元素的大小（字节数）
	closed   uint32         // 通道关闭标志，非零表示通道已关闭
	timer    *timer         // 为该通道提供数据的定时器（用于time.After等场景）
	elemtype *_type         // 通道元素的类型信息
	sendx    uint           // 下一次发送操作的目标索引（环形缓冲区写入位置）
	recvx    uint           // 下一次接收操作的源索引（环形缓冲区读取位置）
	recvq    waitq          // 等待接收操作的goroutine队列
	sendq    waitq          // 等待发送操作的goroutine队列
	bubble   *synctestBubble // 用于同步测试的辅助结构
	lock mutex
}

发送流程#

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
// 处理nil通道
if c == nil {
if !block {
return false
}
gopark(...)
throw("unreachable")
}

// ...调试、race、synctest冗余代码省略

// 无锁快速预判：非阻塞且已满直接提前失败
if !block && c.closed == 0 && full(c) {
return false
}

// ...耗时统计代码省略

lock(&c.lock)

// 通道关闭校验
if c.closed != 0 {
unlock(&c.lock)
panic(plainError("send on closed channel"))
}

// 1.有等待接收G，直接转发数据
if sg := c.recvq.dequeue(); sg != nil {
send(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true
}

// 2.缓冲区有空，写入环形buf
if c.qcount < c.dataqsiz {
qp := chanbuf(c, c.sendx)
// ...内存拷贝、sendx下标轮转、qcount自增细节省略
unlock(&c.lock)
return true
}

// 走到这里：无等待接收G + buf已满
if !block {
unlock(&c.lock)
return false //非阻塞直接发送失败
}

// 3.阻塞逻辑：封装sudog进sendq休眠
gp := getg()
mysg := acquireSudog()
// ...sudog字段赋值细节省略
c.sendq.enqueue(mysg)
gp.parkingOnChan.Store(true)
gopark(chanparkcommit, unsafe.Pointer(&c.lock), ...)

// ...被唤醒后状态校验、资源释放省略
if closed {
panic(plainError("send on closed channel"))
}
return true
}

接收流程#

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
// nil通道处理
if c == nil {
if !block {
return false, false
}
gopark(...)
throw("unreachable")
}

// ...调试、synctest、定时器冗余代码省略

// 非阻塞无锁快速预判：空通道直接提前返回
if !block && empty(c) {
// ...关闭状态校验细节省略
return
}

// ...耗时统计代码省略

lock(&c.lock)

// 通道已关闭
if c.closed != 0 {
if c.qcount == 0 {
unlock(&c.lock)
// 关+空：接收零值，received=false
return true, false
}
// 关闭但buf有数据，继续往下取缓冲数据
} else {
// 1.有阻塞发送G，直接从发送G拿数据
if sg := c.sendq.dequeue(); sg != nil {
recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true, true
}
}

// 2.缓冲区有数据，从环形buf取出
if c.qcount > 0 {
// ...数据拷贝、recvx下标更新、qcount--细节省略
unlock(&c.lock)
return true, true
}

// 走到这里：无等待发送G + buf空
if !block {
unlock(&c.lock)
return false, false //非阻塞接收失败
}

// 3.阻塞逻辑：封装sudog，当前G入recvq休眠
gp := getg()
mysg := acquireSudog()
// ...sudog属性赋值省略
c.recvq.enqueue(mysg)
gp.parkingOnChan.Store(true)
gopark(chanparkcommit, unsafe.Pointer(&c.lock), ...)

// ...唤醒后资源释放、状态校验省略
success := mysg.success
releaseSudog(mysg)
return true, success
}

线程安全#

hchan 自带一把 lock mutex
在发送和接收操作时，会先获取锁，确保线程安全。