教程集 www.jiaochengji.com
教程集 >  Golang编程  >  golang教程  >  正文 golang chan

golang chan

发布时间:2022-03-01   编辑:jiaochengji.com
教程集为您提供golang chan等资源,欢迎您收藏本站,我们将为您提供最新的golang chan资源

chan是我们学习golang绕不开的一个话题,今天我就不讲基础的使用了,因为太多这种文章了,我讲一下channel底层的实现和它的数据结构

<h3>必须了解的数据结构</h3> <pre><code class="lang-go hljs">type hchan struct { qcount uint // 所有数据 dataqsiz uint // 数据size buf unsafe.Pointer // 指向真实数据的指针 elemsize uint16 closed uint32 elemtype *_type // 数据类型 sendx uint // send index recvx uint // receive index recvq waitq // list of recv waiters sendq waitq // list of send waiters // lock protects all fields in hchan, as well as several // fields in sudogs blocked on this channel. // // Do not change another G's status while holding this lock // (in particular, do not ready a G), as this can deadlock // with stack shrinking. lock mutex } //qcount:代表 chan 中已经接收但还没被取走的元素的个数。=len(ch) //dataqsiz:队列的大小。chan 使用一个循环队列来存放元素 // buf:存放元素的循环队列的 buffer。 //elemtype 和 elemsize:chan 元素的类型和元素大小,一般分为普通类型和指针类型,在makechan函数中会判断是否为指针类型 //sendx:处理发送数据的指针在 buf 中的位置。一旦接收了新的数据,指针就会加上 elemsize,移向下一个位置。buf 的总大小是 elemsize 的整数倍,而且 //buf 是一个循环列表。 //recvx:处理接收请求时的指针在 buf 中的位置。一旦取出数据,此指针会移动到下一个位置。 //recvq:chan 是多生产者多消费者的模式,如果消费者因为没有数据可读而阻塞了,就会被加入到 recvq 队列中。 //sendq:如果生产者因为 buf 满了而阻塞,会被加入到 sendq 队列中。 </code></code></pre> <pre><code class="lang-go hljs">type waitq struct { first *sudog last *sudog } // 就是gopark掉了的goroutine队列 //单个节点结构体 type sudog struct { g *g isSelect bool next *sudog prev *sudog elem unsafe.Pointer // data element (may point to stack) acquiretime int64 releasetime int64 ticket uint32 parent *sudog // semaRoot binary tree waitlink *sudog // g.waiting list or semaRoot waittail *sudog // semaRoot c *hchan // channel } </code></code></pre> <h3>在说channel之前你应该了解的</h3> <pre><code class="lang-go hljs">用通俗一点的话来说,channel实际底层并发是靠锁lock实现的,数据写入buf和从buf读出来都有加锁的动作, 存储数据是靠一个循环队列来保存的,队列的大小就是属性buf指向的队列长度,队列长度(len)就是qcount, 实际存储数据是buf的指向的循环队列,然后sendx和recvx是来控制数据读取和删除的,顺序就是按照buf的 索引顺序来的。然后还会保存两个队列:sendq,recvq,这两个参数实际上是绑定wait状态下的g,也就是 被park掉的goroutine,要理解这句话就要有点gmp模型的概念。 </code></code></pre> <h5>1.我们创建一个buffer channel</h5>
image.png
<h5>2.此时的结构体应该是这样的</h5>
image.png
<h5>3.放入一个数据到chan</h5>
image.png
<h5>4.取走一个数据</h5>
image.png
<h5>5.那么简单的存放数据知道了,它是如何将goroutine阻塞的呢---先将一下GMP</h5>

因为这个图是网上扣下来的,不知道原图是谁的,感觉画的很不错


image.png
image.png

这里在做个简单的解释:后面专门会出一个文章来讲解一下GMP调度模型。M是一个内核线程的一个映射(实际上的原理比这个复杂许多,涉及到线程模型,后面写篇文章细讲),P是我们的调度器,一般又称为schedule,正常情况下,我们在代码执行写了一个go func,那么这个go在代码编译的时候就会被丢入一个P的队列中:就像一堆排队的地鼠G去领钱,P就是包工头,第一个地鼠把自己的挖的洞的照片给P看,P看完就给地鼠拍拍灰(准备环境),就让G去M里面去领钱,(领钱就是执行),M呢是个小金库,但是每个地鼠只有10ms时间去领钱,过了就会被p强制的给退出来,有的地鼠不用10ms就能自己退出来

<h4>chan如何将goroutine阻塞的</h4>
image.png

这里需要了解一下gopark:将goroutine进行休眠

<h4>此时GMP应该是这个样子</h4>
image.png

此时G就没有人执行了,所以就看起来像卡住了一样,那么就有个问题了:怎么让它继续运行呢,我们不是将chan读出来goroutine又能继续执行了吗?

<h4>此时Chan的结构体变成了这个样子</h4>
image.png

哪些被游离的gorouine 就被chan的队列给抓住了,放到它的发送或者等待队列中。

<h4>此时这里sendq/recvq长这个样子</h4>
image.png

这里怎么读呢:我就简单解释一下,没这个图,我想自己画一个发现太丑了
此时buf的被别人取走了一个元素,那么就从sendqpop一个出来,是从头部开始pop,然后元素放到了buf,goroutine被设置成runable状态,然后放到P的runable队列中去继续执行下文。
上面是说的发送chan被占满

<h4>如果先来读chan被阻塞,chan内部长什么样子呢</h4>
image.png
<h4>那么chan怎么处理先读后写的这种场景</h4>

按照上面的套路,其实我们可以想,写的时候,先写到buf,然后recvq检测到buf有值了,将buf 的数据pop出来,将goroutine唤醒,数据写入elem指针指向的地址。但是有没有更好的套路呢?

其实根据场景:先有等待者,然后发送者来了,就像去银行存钱一样,取钱的人来了,但是银行没钱,后面存钱的人来了。最大的不同就是gorotuine不需要到chan,就想存钱取钱的人可以不用在银行操作,那么是不是存钱的人就可以直接把钱给取钱的人

实际上的代码也是这么来处理的,就是send 的goroutine直接写了recv的stack,就减少了lock的启动和释放,提高了性能。但是也只有only operations in runtime where this happen:运行时发生这样的情况。

<ul><li>基本上所有的流程就是上面这个图片,下面是比较重要的一些方法,可以按照上面理解去下面看看,不过最好能用goland调试一下,看代码很难看懂,我在看源码的时候,确认chan 数据类型是指针还是普通数据就踩坑了,初次创建的时候调用下面的makechan,我是直接跑到runtime里面去看,发现断点没有走预期路线,我还调试了好久,才发现它的上层是反复调用这个方法,调用了3次,才返回实例,这里面我就不细细的讲了,大家有什么建议也可以评论一下,大家共同学习进步</li> </ul><h3>创建chan:makechan</h3> <pre><code class="lang-go hljs">func makechan(t *chantype, size int) *hchan { elem := t.elem // compiler checks this but be safe. if elem.size >= 1<<16 { //判断元素的大小 throw("makechan: invalid channel element type") } if hchanSize%maxAlign != 0 || elem.align > maxAlign { thro("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")) } // Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers. // buf points into the same allocation, elemtype is persistent. // SudoG's are referenced from their owning thread so they can't be collected. // TODO(dvyukov,rlh): Rethink when collector can move allocated objects. var c *hchan switch { case mem == 0: // Queue or element size is zero. c = (*hchan)(mallocgc(hchanSize, nil, true)) // Race detector uses this location for synchronization. c.buf = c.raceaddr() case elem.ptrdata == 0: // Elements do not contain pointers. // Allocate hchan and buf in one call. c = (*hchan)(mallocgc(hchanSize mem, nil, true)) c.buf = add(unsafe.Pointer(c), hchanSize) default: // Elements contain pointers. 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 } </code></code></pre> <h3>chansend</h3> <pre><code class="lang-go hljs"> func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool { if c == nil { if !block { return false } gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2) throw("unreachable") } if debugChan { print("chansend: chan=", c, "\n") } if raceenabled { racereadpc(c.raceaddr(), callerpc, funcPC(chansend)) } if !block && c.closed == 0 && full(c) { return false } var t0 int64 if blockprofilerate > 0 { t0 = cputicks() } lock(&c.lock) if c.closed != 0 { unlock(&c.lock) panic(plainError("send on closed channel")) } if sg := c.recvq.dequeue(); sg != nil { // Found a waiting receiver. We pass the value we want to send // directly to the receiver, bypassing the channel buffer (if any). send(c, sg, ep, func() { unlock(&c.lock) }, 3) return true } if c.qcount < c.dataqsiz { // Space is available in the channel buffer. Enqueue the element to send. qp := chanbuf(c, c.sendx) if raceenabled { raceacquire(qp) racerelease(qp) } typedmemmove(c.elemtype, qp, ep) c.sendx if c.sendx == c.dataqsiz { c.sendx = 0 } c.qcount unlock(&c.lock) return true } if !block { unlock(&c.lock) return false } // Block on the channel. Some receiver will complete our operation for us. gp := getg() mysg := acquireSudog() mysg.releasetime = 0 if t0 != 0 { mysg.releasetime = -1 } // No stack splits between assigning elem and enqueuing mysg // on gp.waiting where copystack can find it. 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) // Signal to anyone trying to shrink our stack that we're about // to park on a channel. The window between when this G's status // changes and when we set gp.activeStackChans is not safe for // stack shrinking. atomic.Store8(&gp.parkingOnChan, 1) gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2) // Ensure the value being sent is kept alive until the // receiver copies it out. The sudog has a pointer to the // stack object, but sudogs aren't considered as roots of the // stack tracer. KeepAlive(ep) // someone woke us up. if mysg != gp.waiting { throw("G waiting list is corrupted") } gp.waiting = nil gp.activeStackChans = false if gp.param == nil { if c.closed == 0 { throw("chansend: spurious wakeup") } panic(plainError("send on closed channel")) } gp.param = nil if mysg.releasetime > 0 { blockevent(mysg.releasetime-t0, 2) } mysg.c = nil releaseSudog(mysg) return true } </code></code></pre> <h3>close chan</h3> <pre><code class="lang-go hljs"> func closechan(c *hchan) { if c == nil { panic(plainError("close of nil channel")) } lock(&c.lock) if c.closed != 0 { unlock(&c.lock) panic(plainError("close of closed channel")) } if raceenabled { callerpc := getcallerpc() racewritepc(c.raceaddr(), callerpc, funcPC(closechan)) racerelease(c.raceaddr()) } c.closed = 1 var glist gList // release all readers 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 = nil if raceenabled { raceacquireg(gp, c.raceaddr()) } glist.push(gp) } // release all writers (they will panic) for { sg := c.sendq.dequeue() if sg == nil { break } sg.elem = nil if sg.releasetime != 0 { sg.releasetime = cputicks() } gp := sg.g gp.param = nil if raceenabled { raceacquireg(gp, c.raceaddr()) } glist.push(gp) } unlock(&c.lock) // Ready all Gs now that we've dropped the channel lock. for !glist.empty() { gp := glist.pop() gp.schedlink = 0 goready(gp, 3) } } </code></code></pre> <h3>chanrecv方法</h3> <pre><code class="lang-go hljs"> func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) { // raceenabled: don't need to check ep, as it is always on the stack // or is new memory allocated by reflect. if debugChan { print("chanrecv: chan=", c, "\n") } if c == nil { if !block { return } gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2) throw("unreachable") } // Fast path: check for failed non-blocking operation without acquiring the lock. if !block && empty(c) { if atomic.Load(&c.closed) == 0 { return } if empty(c) { // The channel is irreversibly closed and empty. 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 && c.qcount == 0 { if raceenabled { raceacquire(c.raceaddr()) } unlock(&c.lock) if ep != nil { typedmemclr(c.elemtype, ep) } return true, false } if sg := c.sendq.dequeue(); sg != nil { recv(c, sg, ep, func() { unlock(&c.lock) }, 3) return true, true } if c.qcount > 0 { // Receive directly from queue qp := chanbuf(c, c.recvx) if raceenabled { raceacquire(qp) racerelease(qp) } if ep != nil { 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 } // no sender available: block on this channel. gp := getg() mysg := acquireSudog() mysg.releasetime = 0 if t0 != 0 { mysg.releasetime = -1 } // No stack splits between assigning elem and enqueuing mysg // on gp.waiting where copystack can find it. 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) atomic.Store8(&gp.parkingOnChan, 1) gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2) // someone woke us up if mysg != gp.waiting { throw("G waiting list is corrupted") } gp.waiting = nil gp.activeStackChans = false if mysg.releasetime > 0 { blockevent(mysg.releasetime-t0, 2) } closed := gp.param == nil gp.param = nil mysg.c = nil releaseSudog(mysg) return true, !closed } </code></code></pre> 到此这篇关于“ golang chan”的文章就介绍到这了,更多文章或继续浏览下面的相关文章,希望大家以后多多支持JQ教程网!

您可能感兴趣的文章:
golang同步机制之通道
Golang channel 使用指南
golang chan 关闭时的原则
GO chan 编程
golang简易令牌桶算法实现
golang中channal容量的问题
golang 结构体使用chan
Go中使用channel控制goroutine退出
Go语言使用channel等待任务结束
go语言函数传递问题(map、切片、接口、chan)

[关闭]
~ ~