江苏龙网

  1. 主页 > 生活百科 > >

一文搞懂Go通道( 二 )

Go通道

  • buf存放真正的数据
  • sendx和recvx指在环形队列中数据入channel和出channel的位置
  • sendq存放向channel发送数据的goroutine队列
  • recvq存放等待获取channel数据的goroutine队列
  • lock为全局锁
    • 2.3Anwser通过追查到的代码，我们可以回答最开始提出的几个问题了 。
    2.3.1channel为什么是并发安全的？【一文搞懂Go通道】因为做操作之前，都会先获取全局锁，只有获取成功的才能进行操作，保证了并发安全 。
    2.3.2同步通道和异步通道有啥区别？使用的底层数据结构、操作代码都是一样的，只不过dataqsiz的值不一样，一个为0，一个为正数 。
    2.3.3通道为何会阻塞协程？当通道已经满了，但协程继续往通道里写入，或者通道里没有数据，但是协程从通道里获取数据时，协程会被阻塞 。
    实现的原理与Golang并发调度的GMP模型强相关 。
    写入满通道的流程
    1. 当前goroutine（G1）创建自身的一个引用（sudog），放置到hchan的sendq队列
    2. 当前goroutine（G1）会调用gopark函数，将当前协程置为waiting状态；
    3. 将M和G1绑定关系断开；
    4. scheduler会调度另外一个就绪态的goroutine与M建立绑定关系，然后M 会运行另外一个G 。
    读取空通道的流程
    1. 当前goroutine（G2）会创建自身的一个引用（sudog）
    2. 将代表G2的sudog存入recvq等待队列
    3. G2会调用gopark函数进入等待状态，让出OS thread，然后G2进入阻塞态
    2.3.4使用通道导致阻塞的协程是如何解除阻塞的？对于已经满的通道，当有协程G2做读操作时，会解除G1的阻塞，流程为
    1. G2调用 t:=<-ch 获取一个元素A；
    2. 从hchan的buf里面取出一个元素；
    3. 从sendq等待队列里面pop一个sudog；
    4. 将G1要写入的数据复制到buf中A的位置，然后更新buf的sendx和recvx索引值；
    5. G2调用goready(G1)将G1置为Runable状态，表示G1可以恢复运行；
    对于读取空的通道，当有协程G1做写操作时，会解除G2的阻塞，流程为
    1. 将待写入的消息发送给接收的goroutine G2；
    2. G1调用goready(G2) 将G2设置成就绪状态，等待调度；
    2.4实现我们来看一下chan的具体实现
    2.4.1读取数据// chanrecv receives on channel c and writes the received data to ep.// ep may be nil, in which case received data is ignored.// If block == false and no elements are available, returns (false, false).// Otherwise, if c is closed, zeros *ep and returns (true, false).// Otherwise, fills in *ep with an element and returns (true, true).// A non-nil ep must point to the heap or the caller's stack.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.//// After observing that the channel is not ready for receiving, we observe that the// channel is not closed. Each of these observations is a single word-sized read// (first c.sendq.first or c.qcount, and second c.closed).// Because a channel cannot be reopened, the later observation of the channel// being not closed implies that it was also not closed at the moment of the// first observation. We behave as if we observed the channel at that moment// and report that the receive cannot proceed.//// The order of operations is important here: reversing the operations can lead to// incorrect behavior when racing with a close.if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&atomic.Load(&c.closed) == 0 {return}var t0 int64if 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 {// 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}if c.qcount > 0 {// Receive directly from queueqp := 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 = 0if t0 != 0 {mysg.releasetime = -1}// No stack splits between assigning elem and enqueuing mysg// on gp.waiting where copystack can find it.mysg.elem = epmysg.waitlink = nilgp.waiting = mysgmysg.g = gpmysg.isSelect = falsemysg.c = cgp.param = nilc.recvq.enqueue(mysg)goparkunlock(&c.lock, waitReasonChanReceive, traceEvGoBlockRecv, 3)// someone woke us upif mysg != gp.waiting {throw("G waiting list is corrupted")}gp.waiting = nilif mysg.releasetime > 0 {blockevent(mysg.releasetime-t0, 2)}closed := gp.param == nilgp.param = nilmysg.c = nilreleaseSudog(mysg)return true, !closed}复制代码


    推荐阅读


    上一篇：新鲜的迷迭香怎样保存,迷迭香的食用方法

    下一篇：小菊花类品种大全图片,芦荟图片芦荟品种辨别图片大全