go的通信Channel

一、channel是什么

1.一种通信机制

channel是goroutine与goroutine之间数据通信的一种通信机制。一般都是2个g及以上一起工作。

channel与关键字range和select紧密相关。

二、channel的结构

go源码：GitHub - golang/go: The Go programming language

src/runtime/chan.go

type hchan struct {qcount   uint           // total data in the queue 队列中当前元素计数，满了就=dataqsizdataqsiz uint           // size of the circular queue 环形队列大小(缓存大小)buf      unsafe.Pointer // points to an array of dataqsiz elements 指向任意类型的指针elemsize uint16  //元素大小closed   uint32  //是否关闭,0-未关闭，1-已关闭timer    *timer // timer feeding this chan //定时器elemtype *_type // element type //元素类型sendx    uint   // send index //发送索引recvx    uint   // receive index //结束索引recvq    waitq  // list of recv waiters //接收等待队列（<-ch）sendq    waitq  // list of send waiters //发送等待队列（ch<-）// 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 //锁，保护hchan中的所有字段
}type waitq struct { //等待队列，sudog双向链表结构first *sudog //(伪g)表示等待列表中的g，例如在一个通道上用于发送/接收的glast  *sudog //用acquireSudog分配，releaseSudog释放

1.特点

结构体上可以记住channel的一些比如说

(1)lock 锁：操作channel是互斥的。先获取锁，操作channel,释放锁

(2)elemtype类型：创建的时候必须指定类型(大小可指定，如ch := make(chan int,10))

(3)waitq队列：FIFO先进先出队列，即通道，能通过任意类型(unsafe.Pointer)的数据,(sudog)双向链表的g

(4)dataqsiz通道容量：有值=有缓冲通道，没值=无缓冲通道

(5)qcount通道元素计数：当前通道内的元素个数总数

(6)接受和发送：通信，有人发还要有人收，意味必须2个g及以上的成员一起工作

(7)timer定时器：定时可对channel做特殊操作

(8)closed关闭：写(发送)已关闭通道会panic,读(接收)已关闭通道立刻返回：true,false

三、channel的创建

1.make、var

make(chan类型 元素类型，缓冲容量大小),make初始化channel的值是nil

var chan类型 元素类型:var 只是声明，没初始化，直接操作会报错

func Test_2(t *testing.T) {ch1 := make(chan int)       //双向ch11 := make(chan int, 10)  //双向,带缓冲容量10ch2 := make(chan<- int)     //只写ch22 := make(chan int, 10)  //只写,带缓冲容量10ch3 := make(<-chan float64) //只读ch33 := make(chan int, 10)  //只读,带缓冲容量10//go1.17_spec.html//chan T          // can be used to send and receive values of type T//chan<- float64  // can only be used to send float64s//<-chan int      // can only be used to receive intsvar ch4 chan int//通道是引用类型，通道类型的空值是nil。g.Dump(ch1, ch11, ch2, ch22, ch3, ch33, ch4)//打印//<chan int>//<chan int>//<chan<- int>//<chan int>//<<-chan float64>//<chan int>//<chan int>
}

2.chan类型：

chan类型分3种：双向读写、单向写、单向读

ChannelType = ( "chan" | "chan" "<-" | "<-" "chan" )=（双向|单向发送(写)|单向接收(读)）

单向只写通道：操作读会报错。

如使用场景，上下文：src/context/context.go的Context，

type Context interface {
...// a Done channel for cancellation.Done() <-chan struct{}
...}

单向只读通道：操作写会报错

如使用场景，信号量：src/os/signal/signal.go的handlers，

var handlers struct {sync.Mutex// Map a channel to the signals that should be sent to it.m map[chan<- os.Signal]*handler...
}type stopping struct {c chan<- os.Signalh *handler
}

3.元素类型：

任意数据类型，如：int,float64,bool,map...

4.缓冲容量

第二个参数给定数量,如10

无缓冲通道：ch := make(chan int);

有缓冲通道：ch := make(chan int,10)

四、向channel写数据

channel的发送要注意区分【有缓冲容量】和【无缓冲容量】

1.正常发送

默认情况下，发送和接收会一直阻塞着，直到另一方准备好。使用：ch <- i

不管是【有缓冲容量】和【无缓冲容量】的通道，只有有g在接收，其他g就能正常发

func Test_send1(t *testing.T) {ch := make(chan int) //双向//开启goroutine将1~5的数发送到ch中go func() {for i := 1; i <= 5; i++ {fmt.Println("写入ch 元素：", i)ch <- i}close(ch) //写完，关闭通道}()//在主goroutine中从ch中接收值打印for i := range ch {fmt.Println("读取ch 结果：", i)}//写入ch 元素： 1//写入ch 元素： 2//读取ch 结果： 1//读取ch 结果： 2//写入ch 元素： 3//写入ch 元素： 4//读取ch 结果： 3//读取ch 结果： 4//写入ch 元素： 5//读取ch 结果： 5//主goroutine和goroutine读写互斥，相互竞争锁。直到通道关闭主程结束
}

2.异常发送【无缓冲】

没g在接收：无缓冲,其他g不能发，直接阻塞

注意：select是非阻塞发送，会直接返回false

func Test_send2(t *testing.T) {ch := make(chan int) //双向//开启goroutine将1~5的数发送到ch中go func() {for i := 1; i <= 5; i++ {fmt.Println("写入ch 元素：", i)ch <- i}close(ch) //写完，关闭通道}()//写入ch 元素： 1//无缓存通道:没人接收，尝试发送1时，g被阻塞
}

3.异常发送【有缓冲】

没g在接收：有缓冲,其他g能发缓冲容量个数，再阻塞

有缓存通道，容量5，能先发5个，第6个阻塞

func Test_send4(t *testing.T) {ch := make(chan int, 5) //双向//开启goroutine将1~5的数发送到ch中go func() {for i := 1; i <= 10; i++ {fmt.Println("写入ch 元素：", i)ch <- i}close(ch) //写完，关闭通道}()//写入ch 元素： 1//写入ch 元素： 2//写入ch 元素： 3//写入ch 元素： 4//写入ch 元素： 5//写入ch 元素： 6//无缓存通道:没人接收，前5个元素成功发送，尝试发送6时，g才被阻塞
}

4.已关闭发送

已关闭的channel,操作发送会panic

func Test_send3(t *testing.T) {ch := make(chan int) //双向close(ch)            //关闭通道ch <- 1//在主goroutine中从ch中接收值打印for i := range ch {fmt.Println("读取ch 结果：", i)}//panic: send on closed channel
}

5.发送源码：

ch <- i

//发送入口：参数(通道，数据指针)
func chansend1(c *hchan, elem unsafe.Pointer) {chansend(c, elem, true, getcallerpc())
}
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {//(1)如果通道==nil了，并且block==ture,会调用gopark()函数，使当前的g休眠//从而导致了：all goroutines are asleep - deadlock!if c == nil {if !block {return false}gopark(nil, nil, waitReasonChanSendNilChan, traceBlockForever, 2)throw("unreachable")}...lock(&c.lock)if c.closed != 0 {//(2)如果通道关闭了，不允许发送，直接panicunlock(&c.lock)panic(plainError("send on closed channel"))}...if sg := c.recvq.dequeue(); sg != nil { //(3)接收队列有gsend(c, sg, ep, func() { unlock(&c.lock) }, 3)return true}if c.qcount < c.dataqsiz { //(4)接收队列没有g，看有缓冲容量且还没满，往缓冲中发...}...gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceBlockChanSend, 2) //(5)阻塞...
}
//有接收的g，直接发送
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {...sendDirect(c.elemtype, sg, ep)...
}
//将数据直接写入到接收方的执行栈
func sendDirect(t *_type, sg *sudog, src unsafe.Pointer) {...memmove(dst, src, t.Size_)...
}
// memmove从"from"拷贝n个字节到"to"
func memmove(to, from unsafe.Pointer, n uintptr)

发送数据大概过程：

(1)判断channel是否nil,nil且bolck=true,阻塞当前发送g,等待接收方接收数据

(2)获取锁，如果通道关闭了，不允许发送，直接panic

(3)如果当前有接收队列有g在接收，直接调send()函数发送,返回true

(4)接收队列没有g，看有缓冲容量且还没满，往缓冲中发,返回true

(5)找不到接收的g,且缓冲容量也满了，阻塞当前发送的g,等待接收方接收数据

(6)唤醒等待执行的其他g

尤其注意：通道是nil的情况，直接写数据，会发生all goroutines are asleep - deadlock!的情况

func Test_deadlock1(t *testing.T) {// 未初始化的channel,直接写死锁var ch chan intch <- 1//fatal error: all goroutines are asleep - deadlock!// 初始化无缓冲的channel,直接写死锁ch := make(chan int)ch <- 1//fatal error: all goroutines are asleep - deadlock!
}

四、向channel读数据

1.正常读取：

channel读取数据一般用range来遍历读。

也可以用for循环来遍历，channel的recvq和sendq本质是一个双向链表。

跟发送类似，不管是【有缓冲容量】和【无缓冲容量】的通道，只要有g在接收(接收先于发送)，其他g就能正常发

func Test_send1(t *testing.T) {ch := make(chan int) //双向//开启goroutine将1~5的数发送到ch中go func() {for i := 1; i <= 5; i++ {fmt.Println("写入ch 元素：", i)ch <- i}close(ch) //写完，关闭通道}()//在主goroutine中从ch中接收值打印for i := range ch {fmt.Println("读取ch 结果：", i)}//写入ch 元素： 1//写入ch 元素： 2//读取ch 结果： 1//读取ch 结果： 2//写入ch 元素： 3//写入ch 元素： 4//读取ch 结果： 3//读取ch 结果： 4//写入ch 元素： 5//读取ch 结果： 5//主goroutine和goroutine读写互斥，相互竞争锁。直到通道关闭主程结束
}

2.异常读取【死锁】

(1)读完未关闭的channel会造成死锁

func Test_read8(t *testing.T) {ch := make(chan int) //双向//开启goroutine将1~5的数发送到ch中go func() {for i := 1; i <= 5; i++ {fmt.Println("写入ch 元素：", i)ch <- i}//close(ch) //写完，关闭通道}()//在主goroutine中从ch中接收值打印for i := range ch {fmt.Println("读取ch 结果：", i)}//写入ch 元素： 1//写入ch 元素： 2//读取ch 结果： 1//读取ch 结果： 2//写入ch 元素： 3//写入ch 元素： 4//读取ch 结果： 3//读取ch 结果： 4//写入ch 元素： 5//读取ch 结果： 5//fatal error: all goroutines are asleep - deadlock!
}

同样：如果channel没关闭，一直for循环读，读完之后会报死锁的错误。

func Test_read2(t *testing.T) {ch := make(chan int) //双向for i := 1; i < 9; i++ {v, ok := <-chfmt.Println(v, ok)}
}
func Test_read3(t *testing.T) {ch := make(chan int, 5) //双向for i := 1; i < 9; i++ {v, ok := <-chfmt.Println(v, ok)}
}
func Test_read4(t *testing.T) {var ch chan intfor i := 1; i < 9; i++ {v, ok := <-chfmt.Println(v, ok)}
}

3.已关闭读取

思考：会panic吗？

答案是：不会panic,不影响

已关闭通道，如果有值，返回：值,true

已关闭通道，如果没值，返回：0,false

func Test_read5(t *testing.T) {ch := make(chan int) //双向close(ch)for i := range ch {fmt.Println("读取ch 结果：", i)}//0 false//0 false//0 false//0 false//0 false//0 false//0 false//0 false
}func Test_read6(t *testing.T) {ch := make(chan int, 5) //双向for i := 1; i <= 5; i++ {fmt.Println("写入ch 元素：", i)ch <- i}close(ch)for i := 1; i < 9; i++ {v, ok := <-chfmt.Println(v, ok)}//写入ch 元素： 1//写入ch 元素： 2//写入ch 元素： 3//写入ch 元素： 4//写入ch 元素： 5//1 true//2 true//3 true//4 true//5 true//0 false//0 false//0 false
}

4.读取源码：

v <- ch     ,chanrecv1(c *hchan, elem unsafe.Pointer)

v,ok <-ch ,chanrecv2(c *hchan, elem unsafe.Pointer) (received bool)

//接收程序入口
func chanrecv1(c *hchan, elem unsafe.Pointer) {chanrecv(c, elem, true)
}
func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {_, received = chanrecv(c, elem, true)return
}func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {...//(1)如果channel==nil,block==true,调gopart()阻塞当前读取的g//和发送一样，容易造成all g asleep()的问题if c == nil {if !block {return}gopark(nil, nil, waitReasonChanReceiveNilChan, traceBlockForever, 2)throw("unreachable")}...lock(&c.lock)if c.closed != 0 {//(2)通道已关闭，并且没有缓冲值，直接返回true, falseif c.qcount == 0 {...return true, false}} else {if sg := c.sendq.dequeue(); sg != nil {//(3)否则，如果有发送队列在等待，直接调recv进行接收recv(c, sg, ep, func() { unlock(&c.lock) }, 3)return true, true}}//(4)通道未关闭，且缓冲有值，直接读缓冲队列的值if c.qcount > 0 {...return true, true}...gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceBlockChanRecv, 2) //(5)通道没关闭又没有发送的g,且缓冲也是空的，则阻塞当前读取的g...releaseSudog(mysg) //(6)唤醒等待执行的其他g...
}
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {...recvDirect(c.elemtype, sg, ep)...
}
func recvDirect(t *_type, sg *sudog, dst unsafe.Pointer) {...memmove(dst, src, t.Size_)...
}
//memmove从"from"拷贝n个字节到"to"
func memmove(to, from unsafe.Pointer, n uintptr)

（1）如果channel==nil,block==true,调gopart()阻塞当前读取的g

（2）通道已关闭，并且没有缓冲值，直接返回true, false

（3）否则，如果有发送队列在等待，直接调recv进行接收

（4）通道未关闭，且缓冲有值，直接读缓冲队列的值

（5）通道没关闭又没有发送的g,且缓冲也是空的，则阻塞当前读取的g

（6）唤醒等待执行的其他g

读取也要注意：通道是nil的情况，直接读数据，会发生all goroutines are asleep - deadlock!的情况

func Test_deadlock2(t *testing.T) {// 未初始化的channel,直接读死锁var ch chan int<-ch//fatal error: all goroutines are asleep - deadlock!// 初始化无缓冲的channel,直接读死锁ch := make(chan int)val, ok := <-chfmt.Println(val, ok)//  //fatal error: all goroutines are asleep - deadlock!
}

五、channel死锁问题

channel使用不当，很容易造成死锁的问题，死锁的本质是：all goroutines are asleep

1.未关闭channel

通道写完不关闭，容易造成死锁

func Test_deadlock5(t *testing.T) {// 初始化有缓冲的channel,先写后读，读完数据后死锁ch := make(chan int, 5)ch <- 1ch <- 2ch <- 3//range 是阻塞读for v := range ch {fmt.Println(v)}//1//2//3//fatal error: all goroutines are asleep - deadlock!
}

2.已关闭channel

正常使用channel,记得关闭通道

func Test_deadlock5(t *testing.T) {// 初始化有缓冲的channel,先写后读，读完数据后死锁ch := make(chan int, 5)ch <- 1ch <- 2ch <- 3close(ch)//range 是阻塞读for v := range ch {fmt.Println(v)}//1//2//3
}

3.读要先于发 

思考为什么无缓冲 Channel 读要先于发？尝试从源码角度分析，因为无缓存channel的接收方会从发送方栈拷贝数据后，发送方才会被放回调度队列种，等待重新调度，如果一直没有读，发就一直卡住，无法被唤醒

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
...
recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
...
}//1.c通道
//2.发送方sg发送的值被放入通道中，发送方被唤醒，继续它的快乐之路
//3.接收方接收到的值(当前G)为写入ep
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {//无缓冲读if c.dataqsiz == 0 {...if ep != nil {// copy data from sender 接收是直接从发送的栈进行拷贝recvDirect(c.elemtype, sg, ep)}} else {//有缓冲读// 从缓存队列拷贝qp := chanbuf(c, c.recvx)...}gp.param = unsafe.Pointer(sg)...//唤醒g准备执行goready(gp, skip+1)
}func recvDirect(t *_type, sg *sudog, dst unsafe.Pointer) {// dst is on our stack or the heap, src is on another stack.// The channel is locked, so src will not move during this// operation.src := sg.elemtypeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.Size_)//从"from"拷贝n个字节到"to",from是src发送方,to就是dst接受方memmove(dst, src, t.Size_)
}
func memmove(to, from unsafe.Pointer, n uintptr)

4.panic问题

(1)已关闭的通道，再次关闭，会panic

(2)已关闭的通道，写入会panic

五、select

待续

六、总结

1.通道创建类型有3种：双向，只读，只写

2.通道有容量可设：无缓冲通道和有缓冲通道

3.通道读写互斥

4.已关闭通道，写panic,读有值，再关闭panic

5.通道通常需要2个g一起工作