go进阶 -深入理解Channel实现原理

Posted 2023-03-05 hguisu

Go的并发模型已经在https://guisu.blog.csdn.net/article/details/129107148 详细说明。

 1、channel使用详解

---

 1、channel概述

Go的CSP并发模型，是通过goroutine和channel来实现的。

• channel是Go语言中各个并发结构体(goroutine)之前的通信机制。 通俗的讲，就是各个goroutine之间通信的”管道“，有点类似于Linux中的管道。
• Go并发的核心哲学是不要通过共享内存进行通信; 相反，通过沟通分享记忆。

      channel是Go提供goroutine间的通信方式，使用channel可以使多个goroutine之间通信。channel是进程内的通信方式，通过channel传递对象的过程和调用函数时的参数传递行为比较一致，比如也可以传递指针等。

如需跨进程通信，Go建议用分布式系统的方法来解决，如使用Socket或者HTTP等通信协议，Go语言在网络方面也有非常完善的支持。

主要应用场景：

• 数据交流：当作并发的 buffer 或者 queue，解决生产者 - 消费者问题。多个 goroutine 可以并发当作生产者（Producer）和消费者（Consumer）。
• 数据传递：一个goroutine将数据交给另一个goroutine，相当于把数据的拥有权托付出去。
• 信号通知：一个goroutine可以将信号(closing，closed，data ready等)传递给另一个或者另一组goroutine。
• 任务编排：可以让一组goroutine按照一定的顺序并发或者串行的执行，这就是编排功能。
• 锁机制：利用channel实现互斥机制。

 2、channel基本语法

每个channel都有一个特殊的类型，也就是channels可发送数据的类型。一个可以发送int类型数据的channel一般写为chan int。
声明通道：var 通道变量 chan 通道类型：var channame chan ElementType
创建通道：make（chan 数据类型, [缓冲大小]）：

channel跟map类似的在使用之前都需要使用make进行初始化
ch1 := make(chan int, 5) 

 未初始化的channel零值默认为nil，是一种特殊的 chan，对值是 nil 的 chan 的发送接收调用者总是会阻塞。

var ch chan int
fmt.Println(ch) // <nil>

channel的基本用法非常简单，它提供了三种类型，分别为只能接收，只能发送，既能接收也能发送这三种类型：
var channame chan <- ElementType //只写：只能发送ElementType
var channame <- chan ElementType //只读：只能从chan里接收ElementType

var channame chan ElementType //能读能写：既能接收也能发送

我们把既能发送也能接收的chan被称为双向chan，把只能接收或者只能发送的chan称为单向。

而对于close方法只能是发送通道拥有。

箭头总是射向左边的，元素类型总在最右边。

如果箭头指向 chan，就表示可以往 chan 中发送(写)数据；

如果箭头远离 chan，就表示 chan 会往外吐数据，即能从chan里接收(读)数据

channel <- 1 //向channel添加一个值为1
<- channel //从channel取出一个值
a := <- channel //从channel取出一个值并赋值给a
a,b := <- channel //从channel取出一个值赋值给a，如果channel已经关闭或channel没有值，b为false

3、通信机制：

• 成对出现：在通信过程中，传数据channel <- data和取数据<-channel必然会成对出现，因为这边传，那边取，两个goroutine之间才会实现通信。
• 阻塞：不管传还是取，必阻塞，直到另外的goroutine传或者取为止。
• channel仅允许被一个goroutine读写。

1）只能接收数据的chan

package main 

import "fmt"

// a 表示只能接收数据的chan
func goChanA(a <-chan int) 
  b := <-a
  fmt.Println("只能接收数据的channal[a]接收到的数据值为", b)


func main() 
  ch := make(chan int, 2)
  go goChanA(ch)
  // 往ch中写入数据值
  ch <- 2
  time.Sleep(time.Second)

结果：只能接收数据的channal[a]接收到的数据值为 2

2）只能发送数据的chan

package main 

import "fmt

func main() 
  ch := make(chan<- int, 2)
  ch <- 200

往 chan 中发送一个数据使用“ch<-”。
这里的 ch 是 chan int 类型或者是 chan <-int。

3）同步，主协程和子协程之间通信：

func main()
    ch := make(chan int)
    go func() 
        ch <- 996 //向ch添加元素
    ()
    a := <- ch
    fmt.Println(a)
    fmt.Println("程序结束！")

4)、两个子协程的通信

使用channel实现两个goroutine之间通信。

func two() 
    tc := make(chan string)
    ch := make(chan int)
    // 第一个协程
    go func() 
        tc <- "协程A，我在添加数据"
        ch <- 1
    ()
    // 第二个协程
    go func() 
        content := <- tc
        fmt.Printf("协程B，我在读取数据:%s\\n",content)
        ch <- 2
    ()
    <- ch
    <- ch
    fmt.Println("程序结素！")

func main()
    two()

5）、channel仅允许被一个goroutine读写。

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)

4、channel缓冲区

无缓冲通道，make(chan int)，指在接收前没有能力保存任何值的通道，这种类型的通道要求发送goroutine和接收goroutine同时准备好，才能完成发送和接收操作。
有缓冲通道，make(chan int, 2)，指在被接收前能存储一个或者多个值的通道，这种类型的通道并不强制要求goroutine之间必须同时完成发送和接收。

 例子：

package main

import "fmt"

func main() 
	ch1 := make(chan int)
	ch1 <- 5
	rec := <-ch1
	fmt.Println("ch1被接受，程序结束:rec:,", rec)

//fatal error: all goroutines are asleep - deadlock!          

由于ch1没有缓冲区，channel没有缓冲区的话：

只有在有接收方能够接收值的时候才能发送成功，否则会一直处于等待发送的阶段。同理，如果对一个无缓冲通道执行接收操作时，没有任何向通道中发送值的操作那么也会导致接收操作阻塞。

如果想要运行成功那么在发送信息前就应该有另外的协程等待着接收

package main

import (
	"fmt"
	"time"
)

func main() 
	ch1 := make(chan int)
	go receive(ch1)
	ch1 <- 5
	time.Sleep(time.Second)


func receive(ch1 chan int) 
	for 
		select 
		case rec2 := <-ch1:
			fmt.Println("ch1被接受，程序结束:rec:,", rec2)
		
	

//ch1被接受，程序结束:rec:, 5  

 但是如果有缓冲区就能避免程序阻塞，可以将发送的channel放在缓冲区直至有接收方将它接收

向channel添加数据超过缓存，会出现死锁：

func main() 
	ch := make(chan int,3)
	ch <- 1
	//<- ch
	ch <- 1
	ch <- 1
	ch <- 1
	fmt.Println("ok")

 二、使用Select来进行调度

---

select就是用来监听和channel有关的IO操作，当 IO 操作发生时，触发相应的动作。
 

Select 和 swith结构很像，但是select中的case的条件只能是I/O。

Select 的使用方式类似于 switch 语句，它也有一系列 case 分支和一个默认的分支。
每个 case分支会对应一个通道的通信（接收或发送）过程。select 会一直等待，直到其中的某个 case 的通信操作完成时，就会执行该 case分支对应的语句。

具体格式如下：

select 
case <-ch1:
	//...
case rec := <-ch2:
	//...
case ch3 <- 10:
	//...
default:
	//默认操作

select里面case是随机执行的，如果都不满足条件，那么就执行default

select总结：

• 每个case必须是一个I/O操作
• case是随机执行的：如果多个 case 同时满足，select 会随机选择一个执行。
• 如果所有case不能执行，那么会执行default
• 如果所有case不能执行，且没有default，会出现阻塞
• 对于没有 case 的 select 会一直阻塞，可用于阻塞 main 函数，防止退出

实现一个一直接收消息：

func main() 
	ch := make(chan int)
	for i := 1; i <= 10; i++ 
		go func(j int) 
			ch <- j
		(i)
	
	for 
		select 
		case a1 := <- ch:
			fmt.Println(a1)
		default:
		
	

示例2:
 

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)
	default:
		fmt.Println("no data received ")
		time.Sleep(time.Second * 1)
	

运行程序，因为当前时间没有到3s，所以select 选择defult

no data received

修改程序，我们注释掉default，并多执行几次结果为

received the data 5

received the data ok

received the data ok

received the data ok

select语句会阻塞，直到监测到一个可以执行的IO操作为止，而这里goRoutineD和goRoutineE睡眠时间是相同的，都是3s，从输出可看出，从channel中读出数据的顺序是随机的。

再修改代码，goRoutineD睡眠时间改成4s

func goRoutineD(ch chan int, i int) 
   time.Sleep(time.Second * 4)
   ch <- i

此时会先执行goRoutineE，select 选择case msgs := <-chs。
 

三、死锁（deadlock）

---

指两个或两个以上的协程的执行过程中，由于竞争资源或由于彼此通信而造成的一种阻塞的现象。

在非缓冲信道若发生只流入不流出，或只流出不流入，就会发生死锁。

下面是一些死锁的例子

package main

func main() 
   ch := make(chan int)
   ch <- 3

上面情况，向非缓冲通道写数据会发生阻塞，导致死锁。解决办法创建缓冲区 ch := make(chan int，3)

package main
import (
   "fmt"
)

func main() 
   ch := make(chan int)
   fmt.Println(<-ch)

向非缓冲通道读取数据会发生阻塞，导致死锁。 解决办法开启缓冲区，先向channel写入数据。

package main

func main() 
   ch := make(chan int, 3)
   ch <- 3
   ch <- 4
   ch <- 5
   ch <- 6

写入数据超过缓冲区数量也会发生死锁。解决办法将写入数据取走。

死锁的情况有很多这里不再赘述。

还有一种情况，向关闭的channel写入数据，不会产生死锁，产生panic。

package main

func main() 
	ch := make(chan int, 3)
	ch <- 1
	close(ch)
	ch <- 2

解决办法别向关闭的channel写入数据。

四、channel实现原理 

---

1、channel数据结构

channel一个类型管道，通过它可以在goroutine之间发送消息和接收消息。它是golang在语言层面提供的goroutine间的通信方式。通过源代码分析程序执行过程，源码src/runtime/chan.go：

channel结构体hchan：

type hchan struct 
  qcount uint          // 当前队列列中剩余元素个数
  dataqsiz uint        // 环形队列长度，即可以存放的元素个数即缓冲区的大小，即make（chan T，N），N.
  buf unsafe.Pointer   // 环形队列列指针
  elemsize uint16      // 每个元素的⼤⼩
  closed uint32        // 标识关闭状态：表示当前通道是否处于关闭状态。创建通道后，该字段设置为0，即通道打开; 通过调用close将其设置为1，通道关闭。
  elemtype *_type      // 元素类型：用于数据传递过程中的赋值；
  sendx uint           // 队列下标，指示元素写⼊入时存放到队列列中的位置 x
  recvx uint           // 队列下标，指示元素从队列列的该位置读出  
  recvq waitq          // 等待读消息的goroutine队列
  sendq  waitq         // 等待写消息的goroutine队列
  lock mutex           // 互斥锁，chan不允许并发读写
 

type waitq struct 
	first *sudog
	last  *sudog

从数据结构可以看出channel由队列、类型信息、goroutine等待队列组成。

2、实现方式

创建channel 有两种，一种是带缓冲的channel，一种是不带缓冲的channel

// 带缓冲
ch := make(chan Task, 6)
// 不带缓冲
ch := make(chan int)

下图展示了可缓存6个元素的channel底层的数据模型如下图：

func makechan(t *chantype, size int) *hchan 

   elem := t.elem
   ...

• dataqsiz：指向队列的长度为6，即可缓存6个元素
• buf：指向队列的内存，队列中还剩余两个元素
• qcount：当前队列中剩余的元素个数
• sendx：指后续写入元素的位置
• recvx：指从该位置读取数据

等待队列

从channel中读数据，如果channel缓冲区为空或者没有缓冲区，当前goroutine会被阻塞；向channel中写数据，如果channel缓冲区已满或者没有缓冲区，当前goroutine会被阻塞。

被阻塞的goroutine将会被挂在channel的等待队列中：

• 因读阻塞的goroutine会被向channel写入数据的goroutine唤醒
• 因写阻塞的goroutine会被从channel读数据的goroutine唤醒

下面展示了一个没有缓冲区的channel，有几个goroutine阻塞等待数据：

注意，一般情况下recvq和sendq至少有一个为空。只有一个例外，那就是同一个goroutine使用select语句向channel一边写数据一边读数据。

3、向channel写数据

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:<nil> last:<nil>
	sendq runtime.waitq : 
		first:<nil> last:<nil>
	lock runtime.mutex : 
		key:0

 写入数据：ch <- 3，底层hchan数据流程如图

 1、锁定整个通道结构。

2、确定写入：如果recvq队列不为空，说明缓冲区没有数据或者没有缓冲区，此时直接从recvq等待队列中取出一个G（goroutine），并把数据写入，最后把该G唤醒，结束发送过程；

3、如果recvq为Empty，则确定缓冲区是否可用。如果可用，从当前goroutine复制数据写入缓冲区，结束发送过程。

4、如果缓冲区已满，则要写入的元素将保存在当前正在执行的goroutine的结构中，并且当前goroutine将在sendq中排队并从运行时挂起（进入休眠，等待被读goroutine唤醒）。

5、写入完成释放锁。

这里我们要注意几个属性buf、sendx、lock的变化。

3、从channel读取操作

几乎和写入操作相同

func goRoutineA(a <-chan int) 
   val := <-a
   fmt.Println("goRoutineA received the data", val)

底层hchan数据流程如图：

1、先获取channel全局锁

2、如果等待发送队列sendq不为空（有等待的goroutine）：

       1）若没有缓冲区，直接从sendq队列中取出G（goroutine），直接取出goroutine并读取数据，然后唤醒这个goroutine，结束读取释放锁，结束读取过程；

       2）若有缓冲区（说明此时缓冲区已满），从缓冲队列中首部读取数据，再从sendq等待发送队列中取出G，把G中的数据写入缓冲区buf队尾，结束读取释放锁；

3、如果等待发送队列sendq为空（没有等待的goroutine）：

      1）若缓冲区有数据，直接读取缓冲区数据，结束读取释放锁。

      2）没有缓冲区或缓冲区为空，将当前的goroutine加入recvq排队，进入睡眠，等待被写goroutine唤醒。结束读取释放锁。

流程图：

ecvq和

recvq和sendq 结构

recvq和sendq基本上是链表，看起来基本如下：