Golang并发编程之Channel详解
作者:IguoChan 发布时间:2024-05-09 14:58:42
0. 简介
传统的并发编程模型是基于线程
和共享内存的同步访问控制
的,共享数据受锁的保护,线程将争夺这些锁以访问数据。通常而言,使用线程安全
的数据结构会使得这更加容易。Go
的并发原语(goroutine
和channel
)提供了一种优雅的方式来构建并发模型。Go
鼓励在goroutine
之间使用channel
来传递数据,而不是显式地使用锁来限制对共享数据的访问。
Do not communicate by sharing memory; instead, share memory by communicating.
这就是Go
的并发哲学,它依赖CSP(Communicating Sequential Processes)
模型,它经常被认为是Go
在并发编程上成功的关键因素。
如果说goroutine
是Go
语言程序的并发体的话,那么channel
就是他们之间的通信机制,前面的系列博客对goroutine
及其调度机制进行了介绍,本文将介绍一下二者之间的通信机制——channel
。
1. channel数据结构
type hchan struct {
qcount uint // total data in the queue
dataqsiz uint // size of the circular queue
buf unsafe.Pointer // points to an array of dataqsiz elements
elemsize uint16
closed uint32
elemtype *_type // element 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
}
在runtime/chan.go
中,channel
被定义如上,其中:
buf
:是有缓存的channel
持有的,用来存储缓存数据,收个循环链表;dataqsiz
:上述缓存数据的循环链表的最大容量,理解为cap()
;qcount
:上述缓存数据的循环链表的长度,理解为len()
;recvx
和sendx
:表示上述缓存的接收或者发送位置;recvq
和sendq
:分别是接收和发送的goroutine
抽象(sudog
)队列,是个双向链表;lock
:互斥锁,用来保证channel
数据的线程安全。
2. channel创建
func makechan64(t *chantype, size int64) *hchan {
if int64(int(size)) != size {
panic(plainError("makechan: size out of range"))
}
return makechan(t, int(size))
}
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 {
throw("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
}
所有的调用最后都会走到runtime.makechan
函数,函数做的事情比较简单,就是初始化一个runtime.hchan
的对象,和map
一样,channel
对外就是一个指针对象(切片和字符串则不是指针对象,以切片为例,可以参考链接)。可以看到:
如果当前
channel
没有缓存,那么就只会runtime.hchan
分配一段空间;如果当前
channel
中存储的类型不是指针类型,那么会为当前的runtime.hchan
和底层的连续数组分配一块连续的内存空间;其他情况下,那么则为
runtime.hchan
和其缓存各自分配一段内存;
3. 数据发送
// entry point for c <- x from compiled code
//go:nosplit
func chansend1(c *hchan, elem unsafe.Pointer) {
chansend(c, elem, true, getcallerpc())
}
channel
的数据发送会调用runtime.chansend1
函数,而该函数则只是调用了runtime.chansend
函数,该函数比较长,我们一点一点分析:
3.1 空通道的数据发送
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")
}
...
}
可以看到,如果通道是nil
,那么往这个通道中写数据时:
非阻塞写会直接返回(在单
channel
发送+default
分支的select
操作时会调用runtime.selectnbsend
函数,从而会非阻塞写);阻塞写(正常的
ch <- v
)时则会通过gopark
函数让出CPU调度权,阻塞此goroutine
;
3.2 直接发送
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
...
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
}
...
}
可以发现,当channel
被关闭后再发送数据,那么会导致panic
。
如果目标channel
没有关闭,且有已经处于读等待的goroutine
,那么会直接从recvq
中取出最先陷入等待的goroutine
,并通过runtime.send
函数向其发送数据:
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
if raceenabled {
if c.dataqsiz == 0 {
racesync(c, sg)
} else {
// Pretend we go through the buffer, even though
// we copy directly. Note that we need to increment
// the head/tail locations only when raceenabled.
racenotify(c, c.recvx, nil)
racenotify(c, c.recvx, sg)
c.recvx++
if c.recvx == c.dataqsiz {
c.recvx = 0
}
c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
}
}
if sg.elem != nil {
sendDirect(c.elemtype, sg, ep)
sg.elem = nil
}
gp := sg.g
unlockf()
gp.param = unsafe.Pointer(sg)
sg.success = true
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, skip+1)
}
可以看到,以上函数做了两件事:
调用
sendDirect
函数将发送的数据拷贝到接收协程的变量所在的地址上;通过
goready
函数唤醒协程,将其状态置为_Grunnable
后放置到处理器的队列的下一个待处理goroutine
;
3.3 缓存区
如果没有已经处于读等待的goroutine
,且创建的channel
包含缓存,并且缓存还没有满,那么会执行以下代码:
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
...
if c.qcount < c.dataqsiz {
// Space is available in the channel buffer. Enqueue the element to send.
qp := chanbuf(c, c.sendx)
if raceenabled {
racenotify(c, c.sendx, nil)
}
typedmemmove(c.elemtype, qp, ep)
c.sendx++
if c.sendx == c.dataqsiz {
c.sendx = 0
}
c.qcount++
unlock(&c.lock)
return true
}
...
}
在这里会首先通过runtime.chanbuf
函数计算出下一个可以存储的位置,然后通过runtime.typedmemmove
将发送的数据拷贝到缓冲区中并增加sendx
索引和qcount
计数器。等待有接收数据的goroutine
时可以直接从缓存中读取。
3.4 阻塞发送
如果既没有等待读的goroutine
,又没有缓存区或着缓存区满了,那么就会阻塞发送数据:
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
...
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
closed := !mysg.success
gp.param = nil
if mysg.releasetime > 0 {
blockevent(mysg.releasetime-t0, 2)
}
mysg.c = nil
releaseSudog(mysg)
if closed {
if c.closed == 0 {
throw("chansend: spurious wakeup")
}
panic(plainError("send on closed channel"))
}
return true
}
调用
runtime.getg
获取此时发送数据的goroutine
;调用
runtime.acquireSudog
获取sudog
结构并设置相关信息;将上一步获取的
sudog
放到发送等待队列,并且调用gopark
挂起当前协程;等待有接收数据的
goroutine
到来后,即唤醒此goroutine
,然后继续往下走;或者close
了此channel
,导致后续的panic
。
4. 接收数据
4.1 空通道的数据接收
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")
}
...
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
}
...
}
以上是通道接收时的一部分代码,可以看到:
和发送数据一样,如果通道是
nil
,且非阻塞读,则会返回,阻塞读后则会挂起;和发送数据时不一样的是,如果是一个已经关闭的通道,其实是可读的,但是读回的数据都是
零值+false
。
4.2 直接接收
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
...
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
}
...
}
当channel
的sendq
队列中包含处于等待状态的goroutine
时,会取出等待的最早的写数据goroutine
,然后调用runtime.recv
进行发送:
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
if c.dataqsiz == 0 {
if raceenabled {
racesync(c, sg)
}
if ep != nil {
// copy data from sender
recvDirect(c.elemtype, sg, ep)
}
} else {
// Queue is full. Take the item at the
// head of the queue. Make the sender enqueue
// its item at the tail of the queue. Since the
// queue is full, those are both the same slot.
qp := chanbuf(c, c.recvx)
if raceenabled {
racenotify(c, c.recvx, nil)
racenotify(c, c.recvx, sg)
}
// copy data from queue to receiver
if ep != nil {
typedmemmove(c.elemtype, ep, qp)
}
// copy data from sender to queue
typedmemmove(c.elemtype, qp, sg.elem)
c.recvx++
if c.recvx == c.dataqsiz {
c.recvx = 0
}
c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
}
sg.elem = nil
gp := sg.g
unlockf()
gp.param = unsafe.Pointer(sg)
sg.success = true
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, skip+1)
}
该函数会根据是否存在缓存区分别处理:
如果不存在缓存区,则调用
runtime.recvDirect
函数直接将发送goroutine
存储的数据拷贝到目标内存地址中,相当于直接从这个goroutine
中取数据;如果存在缓存区,那么先将缓存区中的数据拷贝到目标内存地址中,然后将
gp
的数据拷贝到缓存区最后,相当于先从缓存队列头部取出数据给接收goroutine
,在从等待发送goroutine
中取出数据到缓存队列尾部,可以看出,此时队列一定是满的。
最后无论哪种情况,都需要调用goready
唤醒gp
。
4.3 从缓存区拿
其实这里的章节名描述并不准确,在4.2中也存在从缓存区拿数据的情况,差别在于:
4.2中缓存队列是满的,且还有发送阻塞等到的
goroutine
;4.3中不存在发送阻塞等到的
goroutine
。
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
...
if c.qcount > 0 {
// Receive directly from queue
qp := chanbuf(c, c.recvx)
if raceenabled {
racenotify(c, c.recvx, nil)
}
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
}
...
}
和发送时一样,如果缓存区有数据,那么从缓存区拷贝数据。
4.4 阻塞接收
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
...
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)
// 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), 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)
}
success := mysg.success
gp.param = nil
mysg.c = nil
releaseSudog(mysg)
return true, success
}
和阻塞发送类似,如果没有等待发送的goroutine
,且没有缓存区或者缓存区没有数据,那这个时候就需要将此接收goroutine
压到recvq
中,并且gopark
挂起,等待唤醒。
5. 关闭
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, abi.FuncPCABIInternal(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 = unsafe.Pointer(sg)
sg.success = false
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 = unsafe.Pointer(sg)
sg.success = false
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)
}
}
关闭通道的代码看上去很长,实际上在处理完一些特殊情况后,就是对发送和接收队列的数据通通使用goready
唤醒。
6. 总结
在Go
中,虽然极力推崇CSP
哲学,推荐大家使用channel
实现共享内存的保护,但是:
在幕后,通道使用锁来序列化访问并提供线程安全性。 因此,通过使用通道同步对内存的访问,你实际上就是在使用锁。 被包装在线程安全队列中的锁。 那么,与仅仅使用标准库
sync
包中的互斥量相比,Go 的花式锁又如何呢? 以下数字是通过使用 Go 的内置基准测试功能,对它们的单个集合连续调用 Put 得出的。
`> BenchmarkSimpleSet-8 3000000 391 ns/op`
`> BenchmarkSimpleChannelSet-8 1000000 1699 ns/o`
就我个人的理解而言:
在进行数据的传输时使用
channel
;在进行内存数据的保护时使用
sync.Mutex
;利用
channel
和select
的特性,实现类似于Linux epoll
的功能。
来源:https://juejin.cn/post/7230348908383453243


猜你喜欢
- 背景在校园里认证上网很麻烦需要web输入账号密码有时还会忘记web地址此时就需要一个人或者程序帮我们实现,这时我想到用python制作这个程
- 本篇文章主要是由于计划使用django写一个计划任务出来,可以定时的轮换值班人员名称或者定时执行脚本等功能,百度无数坑之后,终于可以凑合把这
- 在本文中,我们将介绍 Python 中 _ 字符的不同用法。就像 Python 中的许多其他内容一样,我们会看到 “_” 的不同用法主要是惯
- 随着用户的日益递增,日活和峰值的暴涨,数据库处理性能面临着巨大的挑战。下面分享下对实际10万+峰值的平台的数据库优化方案。与大家一起讨论,互
- 微服务架构中收集通常大家都采用ELK进行日志收集,同时我们还采用了SkyWalking进行链路跟踪,而SkyWalking数据存储也用到了E
- 一、前言阿姨花了30元给幼儿园的小弟弟买了一本习题,里面都是简单的二元加减法。我一听,惊道:“怎么还花钱买题?我动动手指能给你生成一千条。”
- 1 基本信息- 模块主页:[github]- 类型:#第三方库2 安装方法pip install pythonping3 一般使用from
- 三维可视化系统的建立依赖于三维图形平台, 如 OpenGL、VTK、OGRE、OSG等, 传统的方法多采用OpenGL进行底层编程,即对其特
- 1、myisamchk 使用 myisamchk 必须暂时停止 MySQL 服务器。例如,我们要检修 discuz 数据库。执行以下操作:
- 用法:分析一个脚本的依赖: analysis_dependency.py script1.py递归分析依赖: analysis_depend
- multiprocessing多进程概念创建多进程基本流程创建进程对象启动进程 回收进程代码:import multiprocessing
- 1、前不久,friendfeed.com把主导航从上面,移到了右侧。现在,又改到了左侧。2、现在,twitter.com把页签(相当于二级导
- 前两天简单认识了一下PyQt5,通过练习开发了一款在Window下自定义关机的小工具,代码如下:import os,sys,timefrom
- 本文实例为大家分享了JS实现轮播图特效的具体代码,供大家参考,具体内容如下知识点轮播图思想:① 建立一个全局变量索引,始终标记当前显示图片。
- 一个模块可以在逻辑上组织Python代码。将相关的代码到一个模块中,使代码更容易理解和使用。模块是可以绑定和借鉴任意命名属性的P
- CSS浮动一直是个比较让人郁闷的问题,很多的布局问题都出在浮动上,特别是当浮动的列数很多时,但其实只要理解了两列结构的浮动,面对多列数的浮动
- Pytorch torch.distributions库包介绍torch.distributions包包含可参数化的概率分布和采样函数。 这
- 前言分享一下最近在学习Django过程中,遇到和解决的一些有趣的方法和问题一、Django是什么?Django也不用在过多的去介绍了。使用p
- 最近小编遇到一个奇葩问题,就是上传代码时拷贝vs里面的代码不能直接粘贴,否则空格会不符合要求,怎么解决此问题呢?下面小编给大家分享我的解决方
- 前言:python的pandas库中有⼀个⼗分便利的isnull()函数,它可以⽤来判断缺失值,我们通过⼏个例⼦学习它的使⽤⽅法。⾸先我们创