序言
go 中以 goroutine 为最小并发单元,具有比线程更小的开销,使得go的并发和切换性能较强。 而goroutine 之间的相互通信就需要靠 channel来实现。不像 一般语言,线程之间共享信息的方式是通过共享内存方式来进行,
为了保护数据安全,通常访问时候需要加锁,或者采用乐观锁技术(CAS比较),当然go也实现了共享内存加锁方式。 而channel 就像是信道,两个不同的goroutine 相互通信时候,一方充当消费者,另一方充当生产者,信道就像是消息队列
有FIFO的性质,go 能保证channel 的消息不会有丢失的可能。 下面就以channel的使用 和其 内部实现来讲起。
正文
channel 传递的意义
结论:
值得注意的是,channel 设计的本意是在两个goroutine 进行消息传递,如果有多个 goroutine 同时尝试 消费同一个channel 的信息,那么这些goroutine 之间表现一定是竞争关系。
同一个channel在不同的routine 之间共享的话,channel的消息传递消费 是单发模型,点对点的,有多个同时消费的routine则表现为竞争,只有一个routine能被通知到。
联想拓展:
在共享内存并发模型中, 比如 java中,我们对一个 共享变量进行 sychronized 加锁
共享变量: commonLock
Thread A:
sychronized(commonLock){
// action A
}
Thread B:
sychronized(commonLock){
// action B
}
Thread C:
sychronized(commonLock){
// action C
}
|
这里的话,假如A先执行,那么其他 B,C线程进入等待, 当 A执行完, jvm 的调度原则,会自动唤醒 B,C中的其中一个线程继续执行, 如果其中又执行完,那么会继续通知另一个等待线程继续执行。
这里可以看到,在等待一个共享变量锁的同时,这个锁上应该依附了等待的 线程队列信息,每次 离开锁的区域, jvm 会负责调度唤醒另一个 线程进入同步块执行。 这相当于 是 jvm 对于一个共享锁,帮我们进行了队列的逐步唤醒调度。
在这里就不提 object.wait() 释放锁资源,object.notify()通知竞争锁资源等行为了, Thread 之间的 状态,是否休眠,是否可运行,是否等待中,这里也是一个有限状态机,这里先不提。
而回到go 语言的channel ,同样是通知,它的存在意义在于两个gorutine之间的通信, 如果有多个routine ,那么消费者会表现为竞争关系,这其实就不是共享锁 调度队列模型了,那么我们想想go 语言到底帮我们维护了什么呢,我们到底为啥要用这个channel,也就是所谓的CSP并发模型到底简单在哪,在这里给出回答, channel 顾名思义 它是一个管道,是一个双向通道,当有不同的goroutine 投递消息和消费消息的时候,它能保证消息不丢,能被消费到,这个模型其实就是 消息队列中的 生产者 和 消费组的概念
p1 ---m1--> ----> c1
p2 ---m2--> channel ----->c2
p3 ----m3--> ------> c3
p1,p2,p3 三个 goroutine 投递的消息, go保证了消息能投递进channel ,go 保证了 每次 c1,c2,c3 三个goroutine 中的其中一个能消费到其中的一个投递的消息,然后多个消息重复这个流程
( 拓展: 在消息队列概念中, (c1,c2,c3)同属于一个消费组,一个消费组内,每次只有一个消费者能消费,属于单播模式, 同样的消息队列概念里面,如果有多个消费组,那么消费组之间 和生产者关系即为广播关系)
群发模型, 或者是观察者模型,并不是通过一个channel解决的问题,可以通过 channel 数组 来进行。不同goroutine 消费不同的channel,发送信息时候,生产者往 [] channel 里面塞数据。
值得注意的是:当 调用 close(channel)的时候,所有goroutine 都会得到通知 ,va,ok:= ← channel 返回的值 ok 就为 false. 这时候是广播的效用, context 组件就利用了这个 context.Done() 实现了 树形 goroutine 的层级停用。 从 channel 的底层结构里面,channle 也有正在等待 channel信号的goroutine 列表,所以实现广播也是可行的。
- 类似于生产消费者模型,我们很容易产生类比,确实从使用上有确有其类似之处。首先channel 分为 有缓冲区和无缓冲区 两种,表现也不一致。其实也可以统一起来就是,无缓冲区就是缓冲区为0的 channel,其生产者需要依靠消费者的消费,
没有消费者消费的时候,生产者的消息也发不出去。
channel的使用
func main() {
withoutBuffer()
withBuffer()
//selectAndDefaultUse()
//rangeUseChannelConsume()
time.Sleep(time.Minute)
}
func withBuffer() {
defer fmt.Print(enterAndLeaveCall("withBuffer "))
produceAndConsumer(3)
}
func withoutBuffer() {
defer fmt.Print(enterAndLeaveCall("withoutBuffer "))
produceAndConsumer(0)
}
func produceAndConsumer(buffer int) {
tmpChan := make(chan int, buffer)
go func() {
for i := 0; i < 3; i++ {
tmpChan <- i
fmt.Printf("produce %v\n", i)
}
}()
consuFun := func(tag string) {
fmt.Printf("tag %v, and consume %v\n", tag, <-tmpChan)
}
go consuFun("c1")
go consuFun("c2")
time.Sleep(4 * time.Second)
} |
其输出为
enter withoutBuffer <<<<< produce 0 tag c2, and consume 0 tag c1, and consume 1 produce 1 leave withoutBuffer >>>>>> enter withBuffer <<<<< produce 0 produce 1 produce 2 tag c1, and consume 0 tag c2, and consume 1 leave withBuffer >>>>>> |
可以明显看到,如果没有带缓冲区, 生产者会等待有消费者消费的时候,才能继续生产,不然会直接阻塞。
channel的取值还可以通过 select 和 range 等关键字组合
如果是使用 select 可以使用超时机制,防止等待 channel 数据过久
func selectAndDefaultUse() {
tmpChan := make(chan int, 3)
go func() {
for i := 0; i < 10; i++ {
tmpChan <- i
time.Sleep(time.Second)
}
}()
go func() {
for {
select {
case a1 := <-tmpChan:
fmt.Printf("select case value is %v \n", a1)
case <-time.After(300 * time.Millisecond):
fmt.Println("after duration ")
//default:
// fmt.Printf("default case\n ")
// time.Sleep(300 * time.Millisecond)
}
}
}()
}
|
如上, select case 中 加上了一个 time.After 意思是在 这个时间段内如果没有接受到 来自channel的数据,就会走这个分支。而default 也是很明显的意义,就是 如果当下 其他分支channel都还没有数据到达,会直接走default 分支。
range 的使用,直接把channel 里面的数据,通过循环语法糖的方式打印出来
func rangeUseChannelConsume() {
defer fmt.Println(enterAndLeaveCall("rangeUseChannelConsume"))
waitDone := sync.WaitGroup{}
tmpChan := make(chan int, 3)
waitDone.Add(10)
go func() {
for i := 0; i < 10; i++ {
tmpChan <- i
time.Sleep(time.Second)
}
}()
go func() {
for a1 := range tmpChan {
fmt.Println("range use ", a1)
waitDone.Done()
}
}()
waitDone.Wait()
} |
输出为
比较容易理解,在此不表。
enter rangeUseChannelConsume<<<<< range use 0 range use 1 range use 2 range use 3 range use 4 range use 5 range use 6 range use 7 range use 8 range use 9 leave rangeUseChannelConsume>>>>>> |
channel 的内部实现
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
}
type waitq struct {
first *sudog
last *sudog
}
type sudog struct {
// The following fields are protected by the hchan.lock of the
// channel this sudog is blocking on. shrinkstack depends on
// this for sudogs involved in channel ops.
g *g
// isSelect indicates g is participating in a select, so
// g.selectDone must be CAS'd to win the wake-up race.
isSelect bool
next *sudog
prev *sudog
elem unsafe.Pointer // data element (may point to stack)
// The following fields are never accessed concurrently.
// For channels, waitlink is only accessed by g.
// For semaphores, all fields (including the ones above)
// are only accessed when holding a semaRoot lock.
acquiretime int64
releasetime int64
ticket uint32
parent *sudog // semaRoot binary tree
waitlink *sudog // g.waiting list or semaRoot
waittail *sudog // semaRoot
c *hchan // channel
}
|
其中 比较重要的是, 有等待发送的 goroutine 的双向链表 sendq, 等待接收值的goroutine 的双向链表 recvq.
qcount— Channel 中的元素个数;dataqsiz— Channel 中的循环队列的长度;buf— Channel 的缓冲区数据指针;sendx— Channel 的发送操作处理到的位置;recvx— Channel 的接收操作处理到的位置;elemsize 收发元素大小elemtype 收发元素类型sendq 当前 Channel 由于缓冲区空间不足而阻塞的 Goroutine 列表,发送goroutine阻塞recvq 当前 Channel 由于缓冲区空间不足而阻塞的 Goroutine 列表,接收goroutine阻塞
这些等待队列使用双向链表 runtime.waitq 表示,链表中所有的元素都是 runtime.sudog 结构
从channel 调用close(channel)操作函数来看, close channel 会给 收发 goroutine 列表都发送关闭消息
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)
}
} |
简而言之,就是 收消息的goroutine 列表能收到关闭消息, 发送消息的goroutine 会 panic