目录:
- channel 简介
- channel 内部结构
- 创建channel
- 发送数据
- 接收数据
- 关闭channel
- 总结
channel 简介
Go语言有个很出名的话是“以通信的手段来共享内存”，channel就是其最佳的体现，channel提供一种机制，可以同步两个并发执行的函数，还可以让两个函数通过互相传递特定类型的值来通信。
channel有两种初始化方式，分别是带缓存的和不带缓存的:
make(chan int) // 无缓存 chan
make(chan int, 10) // 有缓存 chan
使用方式也很简单:
c := make(chan int)
defer close(c)
go func(){
c <- 5 // send
}()
n := <- c // recv
十分简洁的做到了不同协程的交互。
channel的内部结构
chan的实现在runtime/chan.go，是一个hchan的结构体:
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结构做了一下简单的封装。
创建channel
当我们在代码里面通过make创建一个channel时，实际调用的是下面这个函数:
CALL runtime.makechan(SB)makechan的实现如下所示:
func makechan(t *chantype, size int) *hchan {
elem := t.elem
// 判断 元素类型的大小
if elem.size >= 1<<16 {
throw("makechan: invalid channel element type")
}
// 判断对齐限制
if hchanSize%maxAlign != 0 || elem.align > maxAlign {
throw("makechan: bad alignment")
}
// 判断 size非负 和 是否大于 maxAlloc限制
mem, overflow := math.MulUintptr(elem.size, uintptr(size))
if overflow || mem > maxAlloc-hchanSize || size < 0 {
panic(plainError("makechan: size out of range"))
}
var c *hchan
switch {
case mem == 0: // 无缓冲区，即 make没设置大小
c = (*hchan)(mallocgc(hchanSize, nil, true))
c.buf = c.raceaddr()
case elem.ptrdata == 0: // 数据类型不包含指针
c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
c.buf = add(unsafe.Pointer(c), hchanSize)
default: // 如果包含指针
// Elements contain pointers.
c = new(hchan)
c.buf = mallocgc(mem, elem, true)
}
c.elemsize = uint16(elem.size)
c.elemtype = elem
c.dataqsiz = uint(size)
if debugChan {
print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
}
return c
}
根据上面的代码，我们可以看到，创建channel分为三种情况:
第一种缓冲区大小为0，此时只需要分配hchansize大小的内存就ok
第二种缓冲区大小不为0，且channel的类型不包含指针，此时buf为hchanSize+元素大小*元素个数的连续内存
第三种缓冲区大小不为0，且channel的类型包含指针，则不能简单的根据元素的大小去申请内存，需要通过mallocgc去分配内存
发送数据
发送数据会调用chan.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 <-c: // 这里为非阻塞发送
// do some thing
default:
// do some thing
}
chansend接口会对一些条件做判断
如果向一个为nil的channel发送数据，如果是阻塞发送会一直阻塞：
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 < c.dataqsiz { // 如果 未满
// Space is available in the channel buffer. Enqueue the element to send.
qp := chanbuf(c, c.sendx)
if raceenabled {
raceacquire(qp)
racerelease(qp)
}
typedmemmove(c.elemtype, qp, ep)
c.sendx++
if c.sendx == c.dataqsiz {
c.sendx = 0
}
c.qcount++
unlock(&c.lock)
return true
}
如果此时环形缓冲区数据满了，如果是阻塞发送，此时会把发送方放到sendq队列中。
接收数据
接收数据会调用下面的接口:
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 := <-c: // 这里为非阻塞接收，没有数据直接返回
// do some thing
default:
// do some thing
}
首先chanrecv也会做一些参数校验
如果channel为nil并且是非阻塞模式，直接返回，如果是阻塞模式，永远等待
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中。
关闭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。
总结
综上分析，在使用channel有这么几点要注意
- 确保所有数据发送完后再关闭channel，由发送方来关闭
- 不要重复关闭channel
- 不要向为nil的channel里面发送值
- 不要向为nil的channel里面接收值
- 接收数据时，可以通过返回值判断是否ok
n , ok := <- c
if ok{
// do some thing
}
这样防止channel被关闭后返回了零值，对业务造成影响
更多文章请关注微信公众号: 黑客的成长秘籍