Go channel 实现原理分析

时间 2019-12-07

标签 channel 实现原理分析繁體版

原文原文链接

channel一个类型管道，经过它能够在goroutine之间发送和接收消息。它是Golang在语言层面提供的goroutine间的通讯方式。git

众所周知，Go依赖于称为CSP（Communicating Sequential Processes）的并发模型，经过Channel实现这种同步模式。Go并发的核心哲学是不要经过共享内存进行通讯; 相反，经过沟通分享记忆。github

下面以简单的示例来演示Go如何经过channel来实现通讯。golang

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 received the data 3并发

上面只是个简单的例子，只输出goRoutineA ，没有执行goRoutineB，说明channel仅容许被一个goroutine读写。ide

接下来咱们经过源代码分析程序执行过程，在讲以前，若是不了解go 并发和调度相关知识。请阅读这篇文章ui

https://github.com/guyan0319/...this

说道channel这里不得不提通道的结构hchan。spa

hchan

源代码在src/runtime/chan.go.net

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
}

说明：

qcount uint // 当前队列中剩余元素个数
dataqsiz uint // 环形队列长度，即缓冲区的大小，即make（chan T，N），N.
buf unsafe.Pointer // 环形队列指针
elemsize uint16 // 每一个元素的大小
closed uint32 // 表示当前通道是否处于关闭状态。建立通道后，该字段设置为0，即通道打开; 经过调用close将其设置为1，通道关闭。
elemtype *_type // 元素类型，用于数据传递过程当中的赋值；
sendx uint和recvx uint是环形缓冲区的状态字段，它指示缓冲区的当前索引 - 支持数组，它能够从中发送数据和接收数据。
recvq waitq // 等待读消息的goroutine队列
sendq waitq // 等待写消息的goroutine队列
lock mutex // 互斥锁，为每一个读写操做锁定通道，由于发送和接收必须是互斥操做。

这里sudog表明goroutine。

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

// 带缓冲
ch := make(chan Task, 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:<nil> last:<nil>}
    sendq runtime.waitq : 
        {first:<nil> last:<nil>}
    lock runtime.mutex : 
        {key:0}
}

源代码

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

   elem := t.elem
   ...
}

若是咱们建立一个带buffer的channel，底层的数据模型以下图：

向channel写入数据

ch <- 3

底层hchan数据流程如图

发送操做概要

一、锁定整个通道结构。

二、肯定写入。尝试recvq从等待队列中等待goroutine，而后将元素直接写入goroutine。

三、若是recvq为Empty，则肯定缓冲区是否可用。若是可用，从当前goroutine复制数据到缓冲区。

四、若是缓冲区已满，则要写入的元素将保存在当前正在执行的goroutine的结构中，而且当前goroutine将在sendq中排队并从运行时挂起。

五、写入完成释放锁。

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

流程图

从channel读取操做

几乎和写入操做相同

代码

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

底层hchan数据流程如图

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

读取操做概要

一、先获取channel全局锁

二、尝试sendq从等待队列中获取等待的goroutine，

三、若有等待的goroutine，没有缓冲区，取出goroutine并读取数据，而后唤醒这个goroutine，结束读取释放锁。

四、若有等待的goroutine，且有缓冲区（此时缓冲区已满），从缓冲区队首取出数据，再从sendq取出一个goroutine，将goroutine中的数据存入buf队尾，结束读取释放锁。

五、如没有等待的goroutine，且缓冲区有数据，直接读取缓冲区数据，结束读取释放锁。

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

流程图

recvq和sendq 结构

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

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)
    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

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。

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）

指两个或两个以上的协程的执行过程当中，因为竞争资源或因为彼此通讯而形成的一种阻塞的现象。

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

下面是一些死锁的例子

一、

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
}

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

四、

package main

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

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

死锁的状况有不少这里再也不赘述。

参考：

https://codeburst.io/diving-d...

https://speakerdeck.com/kavya...

https://my.oschina.net/renhc/...