Go channel 实现原理分析
channel一个类型管道，通过它可以在goroutine之间发送和接收消息。它是Golang在语言层面提供的goroutine间的通信方式。Go依赖于成为CSP的并发模型，通过Channel实现这种同步模式。Golang并发的核心哲学是不要通过共享内存进行通信。
下面Go通过channel来实现通信例子：
package main
import (
"fmt"
"time"
)
func goRoutineA(a <-chan int) {
val := <-a
fmt.Println("goRoutineA received the data", val)
}
func goRoutineB(b chan int) {
val := <-b
fmt.Println("goRoutineB received the data", val)
}
func main() {
ch := make(chan int, 3)
go goRoutineA(ch)
go goRoutineB(ch)
ch <- 3
time.Sleep(time.Second * 1)
}
终端显示结果：
~$ goRoutineA recv the data 3
上面的例子只输出goRoutineA信息，没有执行goRoutineB说明channel仅允许被一个goroutine读写。
下面通过源码程序执行过程分析，如果对go并发和调度相关知识不了解，可以 预览这里
首先我们看下通道的结构hchan,源码再src/runtime/chan.go下
type hchan struct {
qcount uint // total data in the queue 当前队列里还剩余元素个数
dataqsiz uint // size of the circular queue 环形队列长度，即缓冲区的大小，即make(chan T,N) 中的N
buf unsafe.Pointer // points to an array of dataqsiz elements 环形队列指针
elemsize uint16 //每个元素的大小
closed uint32 //标识当前通道是否处于关闭状态，创建通道后，该字段设置0，即打开通道；通道调用close将其设置为1，通道关闭
elemtype *_type // element type 元素类型，用于数据传递过程中的赋值
sendx uint // send index 环形缓冲区的状态字段，它只是缓冲区的当前索引-支持数组，它可以从中发送数据
recvx uint // receive index 环形缓冲区的状态字段，它只是缓冲区当前索引-支持数组，它可以从中接受数据
recvq waitq // list of recv waiters 等待读消息的goroutine队列
sendq waitq // list of send waiters 等待写消息的goroutine队列
// 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 //互斥锁，为每个读写操作锁定通道，因为发送和接受必须是互斥操作
}
// sudog 代表goroutine
type waitq struct {
first *sudog
last *sudog
}
创建两种channel类型，一个带缓冲区和一个不带缓冲区的channel
//带缓冲区 ch:=make(chan int, 3) //不带缓冲区 ch:=make(chan int)
带缓冲区
ch:=make(chan int, 3)
创建通道后的缓冲通道结构
hchan struct {
qcount uint : 0
dataqsiz uint : 3
buf unsafe.Pointer : 0xc00007e0e0
elemsize uint16 : 8
closed uint32 : 0
elemtype *runtime._type : &{
size:8
ptrdata:0
hash:4149441018
tflag:7
align:8
fieldalign:8
kind:130
alg:0x55cdf0
gcdata:0x4d61b4
str:1055
ptrToThis:45152
}
sendx uint : 0
recvx uint : 0
recvq runtime.waitq :
{first: last:}
sendq runtime.waitq :
{first: last:}
lock runtime.mutex :
{key:0}
}
源码在$GOPATH/src/runtime/chan.go下：
func makechan(t *chantype, size int) *hchan {
elem := t.elem
...
}
创建一个带有buffer的channel，底层的数据结构模型如图：
image.png
向channel中写入数据
ch <- 3
底层hchan数据流程下如图：
image.png
image.png
发送操作步骤：
- 锁定整个通道结构
- 确定写入。城市recvq从等待队列中等待goroutine，然后将元素直接写入goroutine
- 如果recvq为Empty，则确定缓冲区是否可用，如果可用那么从当前goroutine复制数据到缓冲区中。
- 如果缓冲区已经满了，则要写入的元素将保存在当前执行的goroutine结构中，并且当前goroutine在sendq中并且队列从运行时挂起。
- 写入完成释放锁
- 需要注意的接个属性的变化：buf、sendx、lock
执行流程图如下：
image.png
从channel中读取数据
从channel中读取数据操作几乎和写入操作雷同
func goRoutineA(a <-chan int){
val := <- a
fmt.Println("goRoutineA received the data",val)
}
底层hchan数据流转如下图:
image.png
image.png
- 需要注意的几个属性变化 buf、sendx、recvx、lock
读数据操作如下:
- 先获取channel全局锁
- 尝试sendq等待队列中获取等待的goroutine
- 如果有等待的goroutine，没有缓冲区，取出goroutine并读取数据，然后唤醒这个goroutine，结束读取释放锁
- 如果有等待goroutine，且有缓冲区(缓冲区满了)，从缓冲区队列首取数据，再从sendq取出一个goroutine，将goroutine中的数据存放到buf队列尾，结束读取释放锁。
- 如果没有等待的goroutine，且缓冲区有数据，直接读取缓冲区数据，结束释放锁。
- 如果没有等待的goroutine，且没有缓冲区或者缓冲区为空，将当前goroutine加入到sendq队列，进入睡眠，等待被写入goroutine唤醒，结束读取释放锁。
大概流程如下：
image.png
recvq和sendq结构
recvq和sendq基本上是链表，基本如下：
image.png
select
select 就是用来监听和channel有关的IO操作，当前IO操作发生触发相关动作执行
如下例子：
package main
import (
"fmt"
"time"
)
func goRoutineD(ch chan int, i int) {
time.Sleep(time.Second * 3)
ch <- i
}
func goRoutineE(chs chan string, i string) {
time.Sleep(time.Second * 3)
chs <- i
}
func main() {
ch := make(chan int, 5)
chs := make(chan string, 5)
go goRoutineD(ch, 5)
go goRoutineE(chs, "ok")
select {
case msg := <-ch:
fmt.Println(" received the data ", msg)
case msgs := <-chs:
fmt.Println(" received the data ", msgs)
}
}
多次执行后的结果如下：
received the data 5
received the data ok
received the data ok
received the data ok
select 语句会阻塞，知道监测到一个可执行的IO操作为止，goRoutineD和goRoutineE睡眠时间相同，都是3s，从输出可以看出，从channel中读取数据顺序是随机的。
range
可以持续冲channel中读取数据，一直到channel被关闭，当channel中没有数据是会阻塞当前goroutine，这里阻塞和读channel时阻塞处理机制一样。
例子如下：
package main
import (
"fmt"
"time"
)
func goRoutineD(ch chan int, i int) {
for i := 1; i <= 5; i++{
time.Sleep(time.Second * 1)
ch <- i
}
}
func chanRange(chanName chan int) {
for e := range chanName {
fmt.Printf("Get element from chan: %d\n", e)
if len(chanName) <= 0 { // 如果现有数据量为0，跳出循环
break
}
}
}
func main() {
ch := make(chan int, 5)
go goRoutineD(ch, 5)
chanRange(ch)
}
运行结果如下：
Get element from chan: 1 Get element from chan: 2 Get element from chan: 3 Get element from chan: 4 Get element from chan: 5
死锁(deadlock)
死锁是指两个或者两个以上的协程在执行任务过程中，由于竞争资源或者彼此通信而造成的一种阻塞现象。在非缓冲信道如发生只流入不流出或者只流入出不流入就会发生死锁
死锁例子如下：
//向非缓冲区通道写数据会发生阻塞，导致死锁。解决办法是创建缓冲区 ch:=make(chan int,3)
package main
func main(){
ch:=make(chan int)
ch <- 3
}
向非缓冲区通道读取数据会发生阻塞导致死锁，解决办法开启缓冲区，先向channel中写入数据
package main
import(
"fmt"
)
func main(){
ch:=make(chan int)
fmt.Println(<-ch)
}
写入数据超过缓冲区数量也会发生死锁，解决办法将写入数据取走
package main
func main(){
ch:=make(chan int,3)
ch <- 3
ch <- 4
ch <- 5
ch <- 6
}
向关闭的channel写入数据。解决办法别向关闭的channel写入数据。
package main
func main(){
ch:=make(chan int,3)
ch <- 1
close(ch)
ch <- 2
}
可以参考更多死锁例子：
About The Author
