网站导航
> 文章列表 > Go Channel是什么？

Go Channel是什么？

网友: 文章列表 2024-03-21 21:27:01

Go Channel是什么？

并行、并发

对于并行、并发的区别，在《concurrency in Go》中有这样一句话

concurrency is a property of the code; parallelism is a property of running program.

我认为这句话很有意思，并发是我们代码能保证的，可是并行的事情得交给机器来处理。

channel 与 primitive的选择

switch cond{case 性能关键区域:use primitivescase 试图转移数据所有权:use channelcase 保护结构内部状态:use primitivescase 协调多个逻辑块:use channeldefault:use primitives
}

  • 转移数据所有权：分享一段代码产生的结果给另一段

    数据有所有权，并且一种保证并发程序安全的方式就是确定同一时间只有一个并发上下文拥有数据的所有权。channel帮助我们通过编码意图为channel类型来传递这个概念。

    一个好处就是可以创建可缓冲的channel去实现便宜的内存queue，从而解耦生产者消费者。并且通过channel，可以隐式组合多个并发代码

  • 保护结构内部状态：

    可以对调用者隐藏锁临界区实现细节

    type Counter struct{value intmtx sync.Mutex
}func (c *Counter) Increse {c.mtx.lock()defer c.mtx.Unlock()c.value++
}

  • 协调多个逻辑块

    channel天生比内存访问同步primitive更具有可组合性。将锁分散在对象图中听起来像一场噩梦，但若是channel则是被期望，被鼓励的。

    go channel select语句以及作为队列的功能，可以安全地传递，因此控制软件中突发复杂性会更容易得多。

  • 性能关键区域

    channel实际也是使用了primitive，因此channel会更加慢。不过在此之外，出现性能区域也是暗示着我们需要重组程序

在Concurrent in Go有句这样的话，aim for simplicity, use channels when possible, and treat goroutines like a free resource。这也许就暗示了channel在Go中的独特地位吧！

Channel结构

type hchan struct{// chan元素数量qcount uint// 底层循环数组长度dataqsiz uint// 指向底层缓冲数组指针buf unsafe.Pointer// 元素大小elemsize uint16// 是否被关闭标志closed uint32// 元素类型elemtype *_type// 已发送元素在循环数组中索引sendx uint// 已接收元素在循环数组中的索引recvx uint// 等待接收的goroutine队列recvq waitq// 等待发送的goroutine队列sendq waitq// hchan锁lock mutex
}

从上面结构可以看出，channel是一个拥有锁的双向队列，储存了数据、缓冲、等待接收发送队列等。

waitq是sudog的双向链表

type waitq struct{
first *sudog
last *sudog
}

创建

创建过程就是一个分配和初始化的过程。为有缓存channel分配缓存，以及初始化其他字段

const hchanSize = unsafe.Sizeof(hchan{}) + uintptr(-int(unsafe.Sizeof(hchan{}))&(maxAlign-1))
func makechan(t *chantype, size int64) *hchan {elem:=t.elem// 1. 检查channel size，align// 2. 若元素类型不含指针或者size无缓冲类型，则进行一次内存分配var c *hchanif elem.kind&kindNoPointer!=0 || size==0{// 2.1 分配hchan结构体+元素大小*个数内存c=(*hchan)(mallocgc(hchanSize+unitptr(size)*elem.size,nil,true))// 2.2 若是非指针缓冲型channel，分配缓冲if size>0 && elem.size!=0 {c.buf=add(unsafe.Pointer(c),hchanSize)} else {// 2.3 非缓冲，直接指向chan开始位置// 缓冲型，说明是struct{}之类，因为只会用到接收发送游标，也不影响c.buf=unsafe.Pointer(c)}}else {// 3. 否则进行两次内存分配操作c=new(hchan)c.buf=newarray(elem.size)}c.elemsize=uint16(elem.size)c.elemtype=elemc.dataqsiz=uint(size)return c
}

发送

  1. 若channel为nil，则堵塞当前goroutine
  2. 若channel是非缓冲且等待接收队列没有goroutine，或channel是缓冲已满的缓冲型，则堵塞当前goroutine
  3. 锁住channel，保证线程安全
  4. 在channel关闭时，解锁channel，并panic
  5. 在接收队列有goroutine，则将发送的数据拷贝到接收goroutine，并返回
  6. 若还有缓冲空间，则添加数据到缓冲区，并返回
  7. 在没有缓冲空间，且不需要堵塞的情况下，返回错误
  8. 在没有缓冲空间，且需要堵塞的情况下，将当前goroutine加入到发送等待队列
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {// 如果 channel 是 nilif c == nil {// 不能阻塞，直接返回 false，表示未发送成功if !block {return false}// 当前 goroutine 被挂起gopark(nil, nil, "chan send (nil chan)", traceEvGoStop, 2)throw("unreachable")}// 对于不阻塞的 send，快速检测失败场景// 如果 channel 未关闭且 channel 没有多余的缓冲空间。这可能是：// 1. channel 是非缓冲型的，且等待接收队列里没有 goroutine// 2. channel 是缓冲型的，但循环数组已经装满了元素if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||(c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {return false}// 锁住 channel，并发安全lock(&c.lock)// 如果 channel 关闭了if c.closed != 0 {// 解锁unlock(&c.lock)// 直接 panicpanic(plainError("send on closed channel"))}// 如果接收队列里有 goroutine，直接将要发送的数据拷贝到接收 goroutineif sg := c.recvq.dequeue(); sg != nil {send(c, sg, ep, func() { unlock(&c.lock) }, 3)return true}// 对于缓冲型的 channel，如果还有缓冲空间if c.qcount < c.dataqsiz {// qp 指向 buf 的 sendx 位置qp := chanbuf(c, c.sendx)// 将数据从 ep 处拷贝到 qptypedmemmove(c.elemtype, qp, ep)// 发送游标值加 1c.sendx++// 如果发送游标值等于容量值，游标值归 0if c.sendx == c.dataqsiz {c.sendx = 0}// 缓冲区的元素数量加一c.qcount++// 解锁unlock(&c.lock)return true}// 如果不需要阻塞，则直接返回错误if !block {unlock(&c.lock)return false}// channel 满了，发送方会被阻塞。接下来会构造一个 sudog// 获取当前 goroutine 的指针gp := getg()mysg := acquireSudog()mysg.elem = epmysg.waitlink = nilmysg.g = gpmysg.selectdone = nilmysg.c = cgp.waiting = mysggp.param = nil// 当前 goroutine 进入发送等待队列c.sendq.enqueue(mysg)// 当前 goroutine 被挂起goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)// 从这里开始被唤醒了（channel 有机会可以发送了）if mysg != gp.waiting {throw("G waiting list is corrupted")}gp.waiting = nil// 若被唤醒之后success为false则说明channel被关闭了closed := !mysg.successgp.param = nil// 去掉 mysg 上绑定的 channelmysg.c = nilreleaseSudog(mysg)if closed {if c.closed == 0 {throw("chansend: spurious wakeup")}// 关闭之后唤醒会panicpanic(plainError("send on closed channel"))}return true
}

可以堵塞的goroutine限制数量多少？

理论上没有限制，因为是在链表直接添加的

什么goroutine优先接收到数据？

先等待的先接收

为什么关闭已经堵塞住发送goroutine的channel，会导致发送goroutine报错呢？

因为关闭channel的时候会唤醒所有等待的发送goroutine，并继续执行堵塞之后的逻辑，发现goroutine是被关闭后唤醒的就会panic

接收

接收操作有两个写法，一种带ok，反应channel是否关闭；另一种不带Ok。无论哪种写法最终指向了chanrecv函数

  1. 若为nil channel，在不堵塞情况下，直接返回false；在堵塞情况下挂起当前goroutine
  2. 在非堵塞模式下，非缓冲channel发送队列没有goroutine在等待或者缓冲channel缓冲没有元素的情况下，并且还没有关闭，那么就直接返回false
  3. 锁住channel，保证线程安全
  4. 若channel已经关闭，且缓冲没有元素情况下，解锁，若没有忽略返回值，就接收零值。返回selected为true，ok为false
  5. 等待队列存在goroutine（说明缓冲满或者无缓冲），对于无缓冲直接从发送goroutine复制到接收goroutine；对于有缓冲，从数据队列头接收数据并添加发送者的数据到数据队列尾
  6. 缓冲型且并未满的，直接接收一个，解锁，返回true，true
  7. 缓冲型且满，将接收goroutine添加到等待接收队列中
// chanrecv 函数接收 channel c 的元素并将其写入 ep 所指向的内存地址。
// 如果 ep 是 nil，说明忽略了接收值。
// 如果 block == false，即非阻塞型接收，在没有数据可接收的情况下，返回 (false, false)
// 否则，如果 c 处于关闭状态，将 ep 指向的地址清零，返回 (true, false)
// 否则，用返回值填充 ep 指向的内存地址。返回 (true, true)
// 如果 ep 非空，则应该指向堆或者函数调用者的栈func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {// 省略 debug 内容 …………// 如果是一个 nil 的 channelif c == nil {// 如果不阻塞，直接返回 (false, false)if !block {return}// 否则，接收一个 nil 的 channel，goroutine 挂起gopark(nil, nil, "chan receive (nil chan)", traceEvGoStop, 2)// 不会执行到这里throw("unreachable")}// 在非阻塞模式下，快速检测到失败，不用获取锁，快速返回// 当我们观察到 channel 没准备好接收：// 1. 非缓冲型，等待发送列队 sendq 里没有 goroutine 在等待// 2. 缓冲型，但 buf 里没有元素// 之后，又观察到 closed == 0，即 channel 未关闭。// 因为 channel 不可能被重复打开，所以前一个观测的时候 channel 也是未关闭的，// 因此在这种情况下可以直接宣布接收失败，返回 (false, false)if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&atomic.Load(&c.closed) == 0 {return}// 加锁lock(&c.lock)// channel 已关闭，并且循环数组 buf 里没有元素// 这里可以处理非缓冲型关闭 和 缓冲型关闭但 buf 无元素的情况// 也就是说即使是关闭状态，但在缓冲型的 channel，// buf 里有元素的情况下还能接收到元素if c.closed != 0 && c.qcount == 0 {if raceenabled {raceacquire(unsafe.Pointer(c))}// 解锁unlock(&c.lock)if ep != nil {// 从一个已关闭的 channel 执行接收操作，且未忽略返回值// 那么接收的值将是一个该类型的零值// typedmemclr 根据类型清理相应地址的内存typedmemclr(c.elemtype, ep)}// 从一个已关闭的 channel 接收，selected 会返回truereturn true, false}// 等待发送队列里有 goroutine 存在，说明 buf 是满的// 这有可能是：// 1. 非缓冲型的 channel// 2. 缓冲型的 channel，但 buf 满了// 针对 1，直接进行内存拷贝（从 sender goroutine -> receiver goroutine）// 针对 2，接收到循环数组头部的元素，并将发送者的元素放到循环数组尾部if sg := c.sendq.dequeue(); sg != nil {// Found a waiting sender. If buffer is size 0, receive value// directly from sender. Otherwise, receive from head of queue// and add sender's value to the tail of the queue (both map to// the same buffer slot because the queue is full).recv(c, sg, ep, func() { unlock(&c.lock) }, 3)return true, true}// 缓冲型，buf 里有元素，可以正常接收if c.qcount > 0 {// 直接从循环数组里找到要接收的元素qp := chanbuf(c, c.recvx)// …………// 代码里，没有忽略要接收的值，不是 "<- ch"，而是 "val <- ch"，ep 指向 valif ep != nil {typedmemmove(c.elemtype, ep, qp)}// 清理掉循环数组里相应位置的值typedmemclr(c.elemtype, qp)// 接收游标向前移动c.recvx++// 接收游标归零// 居然不是取模，这什么从什么角度考虑的呢？if c.recvx == c.dataqsiz {c.recvx = 0}// buf 数组里的元素个数减 1c.qcount--// 解锁unlock(&c.lock)return true, true}if !block {// 非阻塞接收，解锁。selected 返回 false，因为没有接收到值unlock(&c.lock)return false, false}// 接下来就是要被阻塞的情况了// 构造一个 sudoggp := getg()mysg := acquireSudog()mysg.releasetime = 0// 待接收数据的地址保存下来mysg.elem = epmysg.waitlink = nilgp.waiting = mysgmysg.g = gpmysg.selectdone = nilmysg.c = cgp.param = nil// 进入channel 的等待接收队列c.recvq.enqueue(mysg)// 将当前 goroutine 挂起goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)// 被唤醒了，接着从这里继续执行一些扫尾工作if mysg != gp.waiting {throw("G waiting list is corrupted")}gp.waiting = nilgp.activeStackChans = falsesuccess := mysg.successgp.param = nilmysg.c = nilreleaseSudog(mysg)return true, success
}

忽略无堵塞模式下的简洁版本

  1. 若为nil channel直接挂起当前goroutine
  2. 若channel已经关闭，且缓冲没有元素情况下，返回select命中、未接收到数据
  3. 等待队列存在goroutine（说明缓冲满或者无缓冲），对于无缓冲直接从发送goroutine复制到接收goroutine；对于有缓冲，从数据队列头接收数据并添加发送者的数据到数据队列尾
  4. 缓冲型且并未满的，直接接收一个，返回select命中、已经收到数据
  5. 缓冲型且满，将接收goroutine添加到等待接收队列中

select接收命中和带ok接收数据是否接收是否在所有情况下都一致呢？

并不是。如果channel被关闭，那么select会认为命中，带ok接收数据会认为未接收

如果有缓冲channel关闭了，那么还能接收到已经发送的数据吗？

能接收到的。

func TestReceiveFromClose(t *testing.T) {ch := make(chan int, 1)go func() {ch <- 1}()time.Sleep(time.Millisecond)close(ch)t.Log(<-ch)
}// 会输出1

从源码也可以看出，在被关闭且无数据时会返回零值，若有数据就能正常接收的

	// 被关闭且无数据时会返回零值if c.closed != 0 && c.qcount == 0 {// ...}// ... // 缓冲型，buf 里有元素，可以正常接收if c.qcount > 0 {// ...}

为什么在发送关闭的channel会panic，而接收关闭的channel则会接收到零值呢？

本人猜测应该是因为作者认为关闭channel是发送者的职责吧

关闭

  1. 若channel为nil，则panic
  2. 若channel已经关闭，panic
  3. 将所有等待接收队列中goroutine释放，并赋予零值，添加到goroutine到链表
  4. 将所有等待发送队列中goroutine释放，添加到goroutine到链表
  5. 唤醒所有添加到链表的goroutine
func closechan(c *hchan) {// nil channel就直接panicif c == nil {panic(plainError("close of nil channel"))}// 锁lock(&c.lock)// 若channel已经关闭就panicif c.closed != 0 {unlock(&c.lock)panic(plainError("close of closed channel"))}// ...// 修改关闭状态c.closed = 1var glist gList// release all readersfor {sg := c.recvq.dequeue()if sg == nil {break}if sg.elem != nil {typedmemclr(c.elemtype, sg.elem)sg.elem = nil}gp := sg.ggp.param = unsafe.Pointer(sg)sg.success = falseglist.push(gp)}// release all writers (they will panic)for {sg := c.sendq.dequeue()if sg == nil {break}sg.elem = nilgp := sg.ggp.param = unsafe.Pointer(sg)sg.success = falseglist.push(gp)}unlock(&c.lock)// Ready all Gs now that we've dropped the channel lock.for !glist.empty() {gp := glist.pop()gp.schedlink = 0goready(gp, 3)}
}

Ref

  1. Concurrency in Go: Tools and Techniques for Developers 1st Edition
  2. https://golang.design/go-questions/channel/struct/
网络标签：

相关问题