GoLangchannel底层代码实现详解

本篇文章向大家介绍《GoLangchannel底层代码实现详解》，主要包括Langchannel、底层，具有一定的参考价值，需要的朋友可以参考一下。

1.channel 简介

Go语言有个很出名的话是“以通信的手段来共享内存”，channel就是其最佳的体现，channel提供一种机制，可以同步两个并发执行的函数，还可以让两个函数通过互相传递特定类型的值来通信

channel有两种初始化方式，分别是带缓存的和不带缓存的:

make(chan int)   // 无缓存 chan
make(chan int, 10)  // 有缓存 chan

使用方式也很简单:

c := make(chan int)
defer close(c)
go func(){
  c 
<p>十分简洁的做到了不同协程的交互。</p>
<h2>2.channel 内部结构</h2>
<p>chan的实现在runtime/chan.go，是一个hchan的结构体:</p>
<pre class="brush:go;">type hchan struct {
  qcount   uint           // 队列中的数据个数
  dataqsiz uint           // 环形队列的大小，channel本身是一个环形队列
  buf      unsafe.Pointer // 存放实际数据的指针，用unsafe.Pointer存放地址，为了避免gc
  elemsize uint16 
  closed   uint32 // 标识channel是否关闭
  elemtype *_type // 数据 元素类型
  sendx    uint   // send的 index
  recvx    uint   // recv 的 index
  recvq    waitq  // 阻塞在 recv 的队列
  sendq    waitq  // 阻塞在 send 的队列
  lock mutex  // 锁 
}

可以看出，channel本身是一个环形缓冲区，数据存放到堆上面，channel的同步是通过锁实现的，并不是想象中的lock-free的方式，channel中有两个队列，一个是发送阻塞队列，一个是接收阻塞队列。当向一个已满的channel发送数据会被阻塞，此时发送协程会被添加到sendq中，同理，当向一个空的channel接收数据时，接收协程也会被阻塞，被置入recvq中。

waitq是一个链表，里面对g结构做了一下简单的封装。

3.创建channel

当我们在代码里面通过make创建一个channel时，实际调用的是下面这个函数:

CALL runtime.makechan(SB)

makechan的实现如下所示:

func makechan(t *chantype, size int) *hchan {
  elem := t.elem
  // 判断 元素类型的大小
  if elem.size >= 1 maxAlign {
    throw("makechan: bad alignment")
  }
  // 判断 size非负 和 是否大于 maxAlloc限制
  mem, overflow := math.MulUintptr(elem.size, uintptr(size))
  if overflow || mem > maxAlloc-hchanSize || size 
<p>根据上面的代码，我们可以看到，创建channel分为三种情况:</p>
<p>1.第一种缓冲区大小为0，此时只需要分配hchansize大小的内存就ok</p>
<p>2.第二种缓冲区大小不为0，且channel的类型不包含指针，此时buf为hchanSize+元素大小*元素个数的连续内存</p>
<p>3.第三种缓冲区大小不为0，且channel的类型包含指针，则不能简单的根据元素的大小去申请内存，需要通过mallocgc去分配内存</p>
<h2>4.发送数据</h2>
<p>发送数据会调用chan.go中的如下接口:</p>
<pre class="brush:go;">CALL runtime.chansend1(SB)

chansend1会调用chansend接口，chansend方法签名如下:

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool

c是具体的channel，ep是发送的数据，block为true表示阻塞的发送，一般向channel发送数据都是阻塞的，如果channel数据满了，会一直阻塞在这里。但是在select中如果有case监听某个channel的发送，那么此时的block参数为false，后续分析select实现会讲到。

select {
    case 
<p>chansend接口会对一些条件做判断</p>
<p>如果向一个为nil的channel发送数据，如果是阻塞发送会一直阻塞：</p>
<pre class="brush:go;"> if c == nil {
    if !block {
      return false
    }
    gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
    throw("unreachable")
  }

首先会加锁，保证原子性，如果向一个已关闭的channel发送数据就会panic。

lock(&c.lock)
if c.closed != 0 {
    unlock(&c.lock)
    panic(plainError("send on closed channel"))
  }

如果此时recvq中有等待协程，就直接调用send函数将数据复制给接收方， 实现如下:

// sg 为接收者协程，ep为发送元素
func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
  if raceenabled {
    if c.dataqsiz == 0 {
      racesync(c, sg)
    } else {
      qp := chanbuf(c, c.recvx)
      raceacquire(qp)
      racerelease(qp)
      raceacquireg(sg.g, qp)
      racereleaseg(sg.g, qp)
      c.recvx++
      if c.recvx == c.dataqsiz {
        c.recvx = 0
      }
      c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
    }
  }
  if sg.elem != nil {
    sendDirect(c.elemtype, sg, ep)
    sg.elem = nil
  }
  gp := sg.g
  unlockf()
  gp.param = unsafe.Pointer(sg)
  if sg.releasetime != 0 {
    sg.releasetime = cputicks()
  }
  goready(gp, skip+1)
}

如果此时没有等待协程，并且数据未满的情况下，就将数据copy到环形缓冲区中，将位置后移一位。

if c.qcount 
<p>如果此时环形缓冲区数据满了，如果是阻塞发送，此时会把发送方放到sendq队列中。</p>
<h2>5.接收数据</h2>
<p>接收数据会调用下面的接口:</p>
<pre class="brush:go;">CALL runtime.chanrecv1(SB)

chanrecv1会调用chanrecv接口，chanrecv方法签名如下:

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool)

c 指需要操作的channel，接收的数据会写到ep中，block与send中的情况一样，表示是阻塞接收还是非阻塞接收，非阻塞接收指在select中case 接收一个channel值:

select {
    case a := 
<p>首先chanrecv也会做一些参数校验</p>
<p>如果channel为nil并且是非阻塞模式，直接返回，如果是阻塞模式，永远等待</p>
<pre class="brush:go;"> if c == nil {
    if !block {
      return
    }
    gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
    throw("unreachable")
  }

随后会加锁，防止竞争读写

lock(&c.lock)

如果向一个已关闭的channel接收数据，此时channel里面还有数据，那么依然可以接收数据，属于正常接收数据情况。

如果向一个已关闭的channel接收数据，此时channel里面没有数据，那么此时返回的是(true,false)，表示有值返回，但不是我们需要的值:

 if c.closed != 0 && c.qcount == 0 {
    if raceenabled {
      raceacquire(c.raceaddr())
    }
    unlock(&c.lock)
    if ep != nil {
      typedmemclr(c.elemtype, ep)  // 将 ep 指向的内存块置 0
    }
    return true, false
  }

接收也分为三种情况：

如果此时 sendq中有发送方在阻塞，此时会调用recv函数:

func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
  if c.dataqsiz == 0 {
    if raceenabled {
      racesync(c, sg)
    }
    if ep != nil {
      recvDirect(c.elemtype, sg, ep)
    }
  } else {
    qp := chanbuf(c, c.recvx)
    if raceenabled {
      raceacquire(qp)
      racerelease(qp)
      raceacquireg(sg.g, qp)
      racereleaseg(sg.g, qp)
    }
    // copy data from queue to receiver
    if ep != nil {
      typedmemmove(c.elemtype, ep, qp)
    }
    // copy data from sender to queue
    typedmemmove(c.elemtype, qp, sg.elem)
    c.recvx++
    if c.recvx == c.dataqsiz {
      c.recvx = 0
    }
    c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
  }
  sg.elem = nil
  gp := sg.g
  unlockf()
  gp.param = unsafe.Pointer(sg)
  if sg.releasetime != 0 {
    sg.releasetime = cputicks()
  }
  goready(gp, skip+1)
}

此时有发送方在等待，表示此时channel中数据已满，这个时候会将channel头部的数据copy到接收方，然后将发送方队列头部的发送者的数据copy到那个位置。这涉及到两次copy操作。

第二种情况是如果没有发送方等待，此时会把数据copy到channel中:

if c.qcount > 0 {
    // Receive directly from queue
    qp := chanbuf(c, c.recvx)
    if raceenabled {
      raceacquire(qp)
      racerelease(qp)
    }
    if ep != nil {
      typedmemmove(c.elemtype, ep, qp)
    }
    typedmemclr(c.elemtype, qp)
    c.recvx++
    if c.recvx == c.dataqsiz {
      c.recvx = 0
    }
    c.qcount--
    unlock(&c.lock)
    return true, true
  }

第三种情况如果channel里面没有数据，如果是非阻塞接收直接返回false，如果是阻塞接收会将接收方协程放入channel的recvq中。

6.关闭channel

关闭channel时会调用如下接口:

func closechan(c *hchan) 

首先会做一些数据校验:

if c == nil {
    panic(plainError("close of nil channel"))
  }
  lock(&c.lock)
  if c.closed != 0 {
    unlock(&c.lock)
    panic(plainError("close of closed channel"))
  }
  if raceenabled {
    callerpc := getcallerpc()
    racewritepc(c.raceaddr(), callerpc, funcPC(closechan))
    racerelease(c.raceaddr())
  }
  c.closed = 1   //置关闭标记位

如果向一个为nil的channel或者向一个已关闭的channel发起close操作就会panic。

随后会唤醒所有在recvq或者sendq里面的协程：

var glist gList
  // release all readers
  for {
    sg := c.recvq.dequeue()
    if sg == nil {
      break
    }
    if sg.elem != nil {
      typedmemclr(c.elemtype, sg.elem)
      sg.elem = nil
    }
    if sg.releasetime != 0 {
      sg.releasetime = cputicks()
    }
    gp := sg.g
    gp.param = nil
    if raceenabled {
      raceacquireg(gp, c.raceaddr())
    }
    glist.push(gp)
  }
  // release all writers (they will panic)
  for {
    sg := c.sendq.dequeue()
    if sg == nil {
      break
    }
    sg.elem = nil
    if sg.releasetime != 0 {
      sg.releasetime = cputicks()
    }
    gp := sg.g
    gp.param = nil
    if raceenabled {
      raceacquireg(gp, c.raceaddr())
    }
    glist.push(gp)
  }
  unlock(&c.lock)

如果存在接收者，将接收数据通过typedmemclr置0。

如果存在发送者，将所有发送者panic。

7.总结

综上分析，在使用channel有这么几点要注意

1.确保所有数据发送完后再关闭channel，由发送方来关闭

2.不要重复关闭channel

3.不要向为nil的channel里面发送值

4.不要向为nil的channel里面接收值

5.接收数据时，可以通过返回值判断是否ok

n , ok := 
<p>这样防止channel被关闭后返回了零值，对业务造成影响</p>
<p>今天关于《GoLangchannel底层代码实现详解》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！</p>