网站导航
> 文章列表 > go进阶(2) -深入理解Channel实现原理

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

网友: 文章列表 2024-03-21 15:49:17

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

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 := <-afmt.Println("只能接收数据的channal[a]接收到的数据值为", b)
}func main() {ch := make(chan int, 2)go goChanA(ch)// 往ch中写入数据值ch <- 2time.Sleep(time.Second)
}

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

2）只能发送数据的chan

package main import "fmtfunc 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 := <- chfmt.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 := <- tcfmt.Printf("协程B，我在读取数据:%s\\n",content)ch <- 2}()<- ch<- chfmt.Println("程序结素！")
}
func main(){two()
}

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

package main
import ("fmt""time"
)
func goRoutineA(a <-chan int) {val := <-afmt.Println("goRoutineA received the data", val)
}
func goRoutineB(b chan int) {val := <-bfmt.Println("goRoutineB  received the data", val)
}
func main() {ch := make(chan int, 3)go goRoutineA(ch)go goRoutineB(ch)ch <- 3time.Sleep(time.Second * 1)
}

6)、一直阻塞的情况

  • 如果当前协程正在从一个没有任何值的通道中读取数据，那么当前协程会阻塞并且等待其他协程往此通道写入值。
  • 因此，读操作将被阻塞。类似的，如果你发送数据到一个通道，它将阻塞当前协程直到有其他协程从通道中读取数据。此时写操作将阻塞 。

主线程在进行通道操作的时候造成死锁：

package mainimport "fmt"func main() {fmt.Println("mainGo start")channel := make(chan string)// 给通道 channel 传入一个数据GoLang.channel <- "GoLang"//此时主线程将阻塞直到有协程接收这个数据. Go的调度器开始调度协程接收通道 channel 的数据// 但是由于没有协程接受，没有协程是可被调度的。所有协程都进入休眠状态，即是主程序阻塞了。fmt.Println("mainGo stop")
}/*
报错
mainGo go start
fatal error: all goroutines are asleep - deadlock!  //所有协程都进入休眠状态,死锁goroutine 1 [chan send]:
main.main()
*/

4、channel缓冲区

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

 例子：

package mainimport "fmt"func main() {ch1 := make(chan int)ch1 <- 5rec := <-ch1fmt.Println("ch1被接受，程序结束:rec:,", rec)
}
//fatal error: all goroutines are asleep - deadlock!

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

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

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

package mainimport ("fmt""time"
)func main() {ch1 := make(chan int)go receive(ch1)ch1 <- 5time.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//<- chch <- 1ch <- 1ch <- 1fmt.Println("ok")
}

  -----

5、 阻塞的 gorutinue 与资源泄露

在 2012 年 Google I/O 大会上，Rob Pike 的 Go Concurrency Patterns 演讲讨论 Go 的几种基本并发模式，如 完整代码 中从数据集中获取第一条数据的函数：

    func First(query string, replicas []Search) Result {c := make(chan Result)replicaSearch := func(i int) { c <- replicas[i](query) }for i := range replicas {go replicaSearch(i)}return <-c}

在搜索重复时依旧每次都起一个 goroutine 去处理，每个 goroutine 都把它的搜索结果发送到结果 channel 中，channel 中收到的第一条数据会直接返回。

返回完第一条数据后，其他 goroutine 的搜索结果怎么处理？他们自己的协程如何处理？

在 First() 中的结果 channel 是无缓冲的，这意味着只有第一个 goroutine 能返回，由于没有 receiver，其他的 goroutine 会在发送上一直阻塞。如果你大量调用，则可能造成资源泄露。

为避免泄露，你应该确保所有的 goroutine 都能正确退出，有 2 个解决方法：

    使用带缓冲的 channel，确保能接收全部 goroutine 的返回结果：

   

func First(query string, replicas ...Search) Result {  c := make(chan Result,len(replicas))    searchReplica := func(i int) { c <- replicas[i](query) }for i := range replicas {go searchReplica(i)}return <-c}

使用 select 语句，配合能保存一个缓冲值的 channel default 语句：

default 的缓冲 channel 保证了即使结果 channel 收不到数据，也不会阻塞 goroutine

    func First(query string, replicas ...Search) Result {  c := make(chan Result,1)searchReplica := func(i int) {select {case c <- replicas[i](query):default:}}for i := range replicas {go searchReplica(i)}return <-c}

    使用特殊的废弃（cancellation） channel 来中断剩余 goroutine 的执行：

   

func First(query string, replicas ...Search) Result {  c := make(chan Result)done := make(chan struct{})defer close(done)searchReplica := func(i int) {select {case c <- replicas[i](query):case <- done:}}for i := range replicas {go searchReplica(i)}return <-c}

Rob Pike 为了简化演示，没有提及演讲代码中存在的这些问题。不过对于新手来说，可能会不加思考直接使用。

 二、使用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 mainimport ("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 mainfunc 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 mainfunc main() {ch := make(chan int, 3)ch <- 3ch <- 4ch <- 5ch <- 6
}

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

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

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

package mainfunc main() {ch := make(chan int, 3)ch <- 1close(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           // 队列下标，指示元素写⼊入时存放到队列列中的位置 xrecvx uint           // 队列下标，指示元素从队列列的该位置读出  recvq waitq          // 等待读消息的goroutine队列sendq  waitq         // 等待写消息的goroutine队列lock mutex           // 互斥锁，chan不允许并发读写
} type waitq struct {first *sudoglast  *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 := <-afmt.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基本上是链表，看起来基本如下：


 



				
				               	

                			

				                

                	

                    	网络标签：                      

                

				
             

		

    

			
    
		

		
	

	     
	

    	

		

    
相关问题