教程集 > Golang编程 > golang教程 > 正文 Golang中select的实现原理

Golang中select的实现原理

发布时间：2022-01-28 编辑：jiaochengji.com

教程集为您提供Golang中select的实现原理等资源，欢迎您收藏本站，我们将为您提供最新的Golang中select的实现原理资源

select是Golang在语言层面提供的多路IO复用的机制。与switch语句稍微有点相似，也会有case和最后的default选择支。每一个case代表一个通信操作（在某个channel上进行发送或者接收）并且会包含一些语句组成一个语句块。

select会等待case中能够执行的case时才去执行。当满足条件时。select才回去通信并执行case之后的语句，这时候其他通信是不执行的。如果有多个case同时就绪，select会随机选择一个执行。一个没有任何case的select语句，会永远等待下去。

<h2>非阻塞select</h2> <pre><code class="lang-go hljs">select { case <-ch: default: } </code></pre> <h2>阻塞select</h2> <pre><code class="lang-go hljs">select { case <-ch: } </code></pre> <blockquote>

注意：对于读channel的case来说，如elem, ok := <-chan1:, 如果channel有可能被其他协程关闭的情况下，一定要检测读取是否成功，因为close的channel也有可能返回，此时ok == false，且不会阻塞

</blockquote> <h1>源码分析</h1> <blockquote>

注意：我会把源码中每个方法的作用都注释出来，可以参考注释进行理解。

</blockquote> <h2>数据结构</h2>

Golang实现select时，定义了一个数据结构表示每个case语句(含defaut，default实际上是一种特殊的case)，select执行过程可以类比成一个函数，函数输入case数组，输出选中的case，然后程序流程转到选中的case块。
我们先看一下case数据结构

runtime\select.go

</blockquote> <pre><code class="lang-go hljs">type scase struct { c *hchan // 当前case语句所操作的channel指针 elem unsafe.Pointer // data element数据元素 kind uint16 //表示该case的类型 pc uintptr // race pc (for race detector / msan) releasetime int64 } </code></pre> <pre><code class="lang-go hljs">//scase.kind values. const ( caseNil = iota caseRecv caseSend caseDefault ) </code></pre>

scase.kind表示该case的类型，分别为：

<ul><li>caseRecv：case语句中尝试读取scase.c中的数据，case <-Chan；</li><li>caseSend：case语句中尝试向scase.c中写入数据，case Chan <- Send；</li><li>caseDefault：default</li></ul><h2>select实现</h2> <pre><code class="lang-go hljs">// cas0为scase数组的首地址，selectgo()就是从这些scase中找出一个返回 // order0指向一个[2 * ncases] uint16类型的数组 // ncases表示scase数组的长度 //返回值: //int ：选中case的编号 //bool：是否成功从channle中读取了数据 func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool) { if debugSelect { print("select: cas0=", cas0, "\n") } cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0)) order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0)) //将cas1从第一个元素开始切片，长度为ncases，容量为ncases //a[x:y:z] 切片长度: y-x 切片容量:z-x scases := cas1[:ncases:ncases] //所有case轮询顺序，占了前面ncase pollorder := order1[:ncases:ncases] //所有case语句中channel序列，占了后面ncase lockorder := order1[ncases:][:ncases:ncases] // 将涉及零个通道的发送/接收案例替换为caseNil，因此以下逻辑可以假定为非nil通道。 for i := range scases { cas := &scases[i] if cas.c == nil && cas.kind != caseDefault { *cas = scase{} } } var t0 int64 if blockprofilerate > 0 { t0 = cputicks() for i := 0; i < ncases; i { scases[i].releasetime = -1 } } // 生成排列顺序，pollorder重新排序 for i := 1; i < ncases; i { j := fastrandn(uint32(i 1)) pollorder[i] = pollorder[j] pollorder[j] = uint16(i) } //通过Hchan地址进行排序以获得锁定顺序 //采用简单的堆排序，以确保n log n个时间和恒定的堆栈占用量 for i := 0; i < ncases; i { j := i // 替换相同channel 的case，以达到去重防止对channel加锁时重复加锁的目的 c := scases[pollorder[i]].c for j > 0 && scases[lockorder[(j-1)/2]].c.sortkey() < c.sortkey() { k := (j - 1) / 2 lockorder[j] = lockorder[k] j = k } lockorder[j] = pollorder[i] } for i := ncases - 1; i >= 0; i-- { o := lockorder[i] c := scases[o].c lockorder[i] = lockorder[0] j := 0 for { k := j*2 1 if k >= i { break } if k 1 < i && scases[lockorder[k]].c.sortkey() < scases[lockorder[k 1]].c.sortkey() { k } if c.sortkey() < scases[lockorder[k]].c.sortkey() { lockorder[j] = lockorder[k] j = k continue } break } lockorder[j] = o } if debugSelect { for i := 0; i 1 < ncases; i { if scases[lockorder[i]].c.sortkey() > scases[lockorder[i 1]].c.sortkey() { print("i=", i, " x=", lockorder[i], " y=", lockorder[i 1], "\n") throw("select: broken sort") } } } //锁住所有的channel sellock(scases, lockorder) var ( gp *g sg *sudog c *hchan k *scase sglist *sudog sgnext *sudog qp unsafe.Pointer nextp **sudog ) loop: // pass 1 - look for something already waiting // 按照随机顺序检测scase中的channel是否ready var dfli int var dfl *scase var casi int var cas *scase var recvOK bool //开始遍历case数组了 for i := 0; i < ncases; i { casi = int(pollorder[i]) cas = &scases[casi] c = cas.c switch cas.kind { case caseNil: continue // 接收chan case caseRecv: sg = c.sendq.dequeue() // 当chan的等待写队列不为空，需要等待 if sg != nil { goto recv } //当chan的缓存队列存在元素时，不需要等待 if c.qcount > 0 { goto bufrecv } // 当chan关闭时 if c.closed != 0 { goto rclose } //发送chan case caseSend: if raceenabled { racereadpc(c.raceaddr(), cas.pc, chansendpc) } // 当chan关闭时 if c.closed != 0 { goto sclose } //当chan的等待读消息的队列不为空，需要等待 sg = c.recvq.dequeue() if sg != nil { goto send } // chan的缓存队列的元素少于缓存容量时，还有位置，不需要等待 if c.qcount < c.dataqsiz { goto bufsend } case caseDefault: dfli = casi dfl = cas } } if dfl != nil { selunlock(scases, lockorder) casi = dfli cas = dfl goto retc } // pass 2 - enqueue on all chans //所有case都未ready，且没有default语句 //将当前协程加入到所有channel的等待队列 gp = getg() if gp.waiting != nil { throw("gp.waiting != nil") } nextp = &gp.waiting for _, casei := range lockorder { casi = int(casei) cas = &scases[casi] if cas.kind == caseNil { continue } c = cas.c sg := acquireSudog() sg.g = gp sg.isSelect = true // No stack splits between assigning elem and enqueuing // sg on gp.waiting where copystack can find it. sg.elem = cas.elem sg.releasetime = 0 if t0 != 0 { sg.releasetime = -1 } sg.c = c // Construct waiting list in lock order. *nextp = sg nextp = &sg.waitlink switch cas.kind { case caseRecv: // 加入等待接收队列 c.recvq.enqueue(sg) case caseSend: // 加入等待发送队列 c.sendq.enqueue(sg) } } // wait for someone to wake us up //当将协程转入阻塞，等待被唤醒 gp.param = nil gopark(selparkcommit, nil, waitReasonSelect, traceEvGoBlockSelect, 1) sellock(scases, lockorder) gp.selectDone = 0 sg = (*sudog)(gp.param) gp.param = nil // pass 3 - dequeue from unsuccessful chans // otherwise they stack up on quiet channels // record the successful case, if any. // We singly-linked up the SudoGs in lock order. //唤醒后返回channel对应的case index casi = -1 cas = nil sglist = gp.waiting // Clear all elem before unlinking from gp.waiting. for sg1 := gp.waiting; sg1 != nil; sg1 = sg1.waitlink { sg1.isSelect = false sg1.elem = nil sg1.c = nil } gp.waiting = nil for _, casei := range lockorder { k = &scases[casei] if k.kind == caseNil { continue } if sglist.releasetime > 0 { k.releasetime = sglist.releasetime } if sg == sglist { // sg has already been dequeued by the G that woke us up. casi = int(casei) cas = k } else { c = k.c if k.kind == caseSend { c.sendq.dequeueSudoG(sglist) } else { c.recvq.dequeueSudoG(sglist) } } sgnext = sglist.waitlink sglist.waitlink = nil //释放所有的锁 releaseSudog(sglist) sglist = sgnext } //没找到case，重新循环 if cas == nil { goto loop } c = cas.c if debugSelect { print("wait-return: cas0=", cas0, " c=", c, " cas=", cas, " kind=", cas.kind, "\n") } if cas.kind == caseRecv { recvOK = true } if raceenabled { if cas.kind == caseRecv && cas.elem != nil { raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc) } else if cas.kind == caseSend { raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc) } } if msanenabled { if cas.kind == caseRecv && cas.elem != nil { msanwrite(cas.elem, c.elemtype.size) } else if cas.kind == caseSend { msanread(cas.elem, c.elemtype.size) } } selunlock(scases, lockorder) goto retc bufrecv: // 可以从缓冲区接收 if raceenabled { if cas.elem != nil { raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc) } raceacquire(chanbuf(c, c.recvx)) racerelease(chanbuf(c, c.recvx)) } if msanenabled && cas.elem != nil { msanwrite(cas.elem, c.elemtype.size) } recvOK = true qp = chanbuf(c, c.recvx) if cas.elem != nil { // 将chan缓存中的数据拷贝到 case.elem。 eg: a := <-ch, a就是case.elem typedmemmove(c.elemtype, cas.elem, qp) } typedmemclr(c.elemtype, qp) c.recvx if c.recvx == c.dataqsiz { c.recvx = 0 } c.qcount-- selunlock(scases, lockorder) goto retc bufsend: //可以发送到缓冲区 if raceenabled { raceacquire(chanbuf(c, c.sendx)) racerelease(chanbuf(c, c.sendx)) raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc) } if msanenabled { msanread(cas.elem, c.elemtype.size) } // 将cas.elem拷贝到chan的缓存中。eg: ch <- a, a 就是 cas.elem typedmemmove(c.elemtype, chanbuf(c, c.sendx), cas.elem) c.sendx if c.sendx == c.dataqsiz { c.sendx = 0 } c.qcount selunlock(scases, lockorder) goto retc recv: //可以从休眠的发件人（sg）接收 recv(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2) if debugSelect { print("syncrecv: cas0=", cas0, " c=", c, "\n") } recvOK = true goto retc rclose: //在封闭channel的末尾读取 selunlock(scases, lockorder) recvOK = false if cas.elem != nil { typedmemclr(c.elemtype, cas.elem) } if raceenabled { raceacquire(c.raceaddr()) } goto retc send: //可以发送到休眠的接收器（sg） if raceenabled { raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc) } if msanenabled { msanread(cas.elem, c.elemtype.size) } send(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2) if debugSelect { print("syncsend: cas0=", cas0, " c=", c, "\n") } goto retc retc: if cas.releasetime > 0 { blockevent(cas.releasetime-t0, 1) } return casi, recvOK sclose: //在关闭的channel上发送 selunlock(scases, lockorder) panic(plainError("send on closed channel")) } </code></pre>

这个函数看起来很复杂，我们总结一下上面的过程：

<ol><li>初始化pollorder切片和lockorder切片，其中pollorder是存放所有case的，lockorder是存放所有channel</li><li>对pollorder进行重排序，打乱所有case的顺序</li><li>锁定scase语句中所有的channel</li><li>按照随机顺序检测pollorder中case对应的channel是否ready：
4.1 如果case可读，则读取channel中数据，解锁所有的channel，然后返回(case index, true)
4.2 如果case可写，则将数据写入channel，解锁所有的channel，然后返回(case index, false)
4.3 所有case都未ready，则解锁所有的channel，然后返回（default index, false）</li><li>所有case都未ready，且没有default语句
5.1 将当前协程加入到所有channel的等待队列
5.2 当将协程转入阻塞，等待被唤醒</li><li>唤醒后返回channel对应的case index
6.1 如果是读操作，解锁所有的channel，然后返回(case index, true)
6.2 如果是写操作，解锁所有的channel，然后返回(case index, false)</li></ol>

其中，很多涉及到channel代码部分可参考之前的文章: Golang中channel的实现原理

<h1>总结</h1> <ul><li>select语句中除default外，各case执行顺序是随机的</li><li>select语句中如果没有default语句，则会阻塞等待任意一个case</li><li>select语句中读操作要判断是否成功读取，关闭的channel也可以读取</li><li>select语句中除default外，每个case只能操作一个channel，要么读要么写</li></ul>

<blockquote class="layui-elem-quote" style="width: 100%;overflow:hidden"> 作者： xzw12138
链接： https://blog.csdn.net/xzw12138/article/details/108104603
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
</blockquote>

到此这篇关于“Golang中select的实现原理”的文章就介绍到这了,更多文章或继续浏览下面的相关文章，希望大家以后多多支持JQ教程网！

您可能感兴趣的文章：
golang url 收集
 golang的select实现原理剖析
 golang中timer定时器实现原理
 golang的协程原理
 Golang协程原理（一）
go 协程
 想系统学习GO语言(Golang
golang中的mmap使用
 golang学习之--select--case 原理【转载】
Golang channel 使用指南

上一篇：golang 网络编程（1）下一篇：Golang - 访问数据库报错后程序继续运行

[关闭]

Golang中select的实现原理

最近更新

浏览排行