我报名参与金石计划1期挑战——分割10万奖池，这是我的第1篇文章，点击查看活动详情
前言
go言语并发哲学
Do not communicate by sharing memory; instead , sharing memory of communicating
不要经过同享内存来通讯，而要经过通讯来实现内存同享。
这便是Go言语的并发哲学，他依赖于CSP，依据channel实现。
什么是CSP
CSP全称是Communicating Sequential Processes，这是Tony Hoare在1978年宣布的ACM中的一篇论文。论文中指出一门编程言语应该注重input和output的原语，尤其是并发变成的代码。
Go是第一个将CSP这些思想引进并发扬光大的言语 。 仅管理内存同步拜访操控在某些状况下大有用途，Go也有相应的sync包进行支撑。大多数的编程言语是依据线程和内存同步拜访操控，Go的并发的模型则是运用goroutine和channel代替。
Go routine解放了程序员，让咱们不用去考虑线程库、线程开支、线程调度等等繁琐的底层问题，goroutine天生就帮咱们解决了。
本文要点是剖析channel的源码层次的内容，gorutine相关的内容能够自行去翻阅资料。本文一切源码的版本为go 1.17.5。
其他阐明
当时在很多公众号和博文中都能找到channel相关的源码解读，我这儿从头写一篇文章来剖析，为的仅仅自己愈加深化的了解和学习，看他人的文章而不动手只能停留在概念层次，只有自己亲身经历了进程，印象才会深刻。
channel的源码结构剖析
channel的底层数据结构是由hchan来实现的。
代码方位
src/runtime/chan.go:hchan
hchan的数据结构
type hchan struct {
qcount uint //当时行列中剩下元素的个数
datasiz uint //环形链的长度
buf unsafe.Pointer //环形行列指针
elemsize uint16 //每个元素的巨细
closed uint32 //标识封闭状况
elemtype *_type //元素类型
sendx uint //已发送元素在循环数组中的索引
recvx uint //已接纳元素在循环数组中的索引
recq waitq //等候接纳的goroutine行列
sendq waitq //等候发送的goroutine行列
lock mutex //锁
}
相关的字段阐明现已在界说中添加了注释阐明，需求要点说的是如下字段
- buf ：指向底层循环数组，只有缓冲型channel才有。
- elemtype：代表元素类型，用于数据传递进程的中赋值
- elemsize：元素巨细，用于在buf中定位元素呈现的方位
- sendx、recvx：均指向底层循环数组，标明当时能够发送和接纳的元素方位索引。
- sendq、recvq：分别标明被堵塞的goroutine，这些goroutine因为测验读取channel中的数据或许向channel发送数据而被堵塞
- lock 用于确保每个读channel和写channel的操作是原子的。
waitq是sudog类型的双向链表，而sudog实际是对goroutine的封装，其结构如下
type waitq struct {
first *sudog
last *sudog
}
channel的操作
channel的操作分为创立、发送、接纳、封闭等4个操作，下面就从源码和流程方面分开展现
创立chan
通道一般有两个方向，发送和接纳。理论上，咱们能够创立一个只发送或许只接纳。在go中经过make来创立channel，代码如下
//无缓冲的通道
ch1 := make(chan int)
//有缓冲的通道
ch2 := make(chan int , 1)
在底层，channel的创立是由makechan函数来履行的，函数界说如下
func makechan(t *chantype, size int64) *hchan
入参包括两个
- t ：类型，对应chantype类型的数据结构
- size：巨细，对应是否为有缓冲仍是无缓冲的channel，值大于零标明是有缓冲的channel，等于0标明无缓冲的channel
创立chan的进程
- 检查元素的巨细，hchan的size以及align等相关信息
- 判断内存巨细是否为0，若为0，则为chan分配内存空间，并将chan的buf地址进行初始化。
- 若元素中不包括指针，则调用mallocgc函数进行内存分配，并将chan的buf指向新的地址，新的地址是依据hchan的当时地址加上hchansize的值最终得到的。
- 若元素中包括指针，则调用new函数来创立hchan指针，并为chan的buf分配内存。
makechan函数代码
//相关的常量界说
const (
maxAlign = 8
hchanSize = unsafe.Sizeof(hchan{}) + uintptr(-int(unsafe.Sizeof(hchan{}))&(maxAlign-1))
debugChan = false
)
func makechan(t *chantype, size uint64) *hchan {
elem := t.elem
// 检测channel的巨细是否现已超过限制，不然抛出反常
if elem.size >= 1<<16 {
throw("makechan: invalid channel element type")
}
//检测元素的align是否超出最大的align，不然抛出反常
if hchanSize%maxAlign != 0 || elem.align > maxAlign {
throw("makechan: bad alignment")
}
//检测溢出状况，不然抛出反常
mem, overflow := math.MulUintptr(elem.size, uintptr(size))
if overflow || mem > maxAlloc-hchanSize || size < 0 {
panic(plainError("makechan: size out of range"))
}
// 当存储在buf中的元素不包括指针时分，hchan不包括指针。
//buf指向相同的内存空间，elemtype是持久的
// Sudog是从从他们自己的线程中引用的，所以不能被收集。
var c *hchan
switch {
case mem == 0:
// 假如行列或许chan size巨细为0，则调用mallocgc函数为chan分配内存空间
c = (*hchan)(mallocgc(hchanSize, nil, true))
// Race detector uses this location for synchronization.
c.buf = c.raceaddr()
case elem.ptrdata == 0:
// 元素中不包括指针，在一次调用中为buf和和hchan分配内存
c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
c.buf = add(unsafe.Pointer(c), hchanSize)
default:
// 元素中包括指针，则运用new关键值来分配内存
c = new(hchan)
c.buf = mallocgc(mem, elem, true)
}
c.elemsize = uint16(elem.size)
c.elemtype = elem
c.dataqsiz = uint(size)
lockInit(&c.lock, lockRankHchan)
if debugChan {
print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
}
return c
}
//chan的raceaddr办法代码如下
func (c *hchan) raceaddr() unsafe.Pointer() {
return unsafe.Pointer(&c.buf)
}
chan中一切分配内存的函数都是有mallcogc函数来完结的，其源代码在src/runtime/malloc.go:mallocgc中，感兴趣的能够去查看下。
往channel中发送数据
往channel的发送数据由chansend1和chansend两个办法完结，chansend1底层也是直接调用chansend办法完结发送的
chansend函数源代码
// entry point for c <- x from compiled code
//go:nosplit
func chansend1(c *hchan, elem unsafe.Pointer) {
chansend(c, elem, true, getcallerpc())
}
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
//假如channel为nil
if c == nil {
//不堵塞，直接回来false，标明没有发送成功
if !block {
return false
}
//将当时goroutine挂起，并抛出unreachable反常
gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
throw("unreachable")
}
//debug形式下打印chan相关的信息
if debugChan {
print("chansend: chan=", c, "\n")
}
//race相关的信息
if raceenabled {
racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
}
//对于非堵塞的 而且chan的closed标记为0 而且chan的行列现已满了，直接回来false
if !block && c.closed == 0 && full(c) {
return false
}
var t0 int64
if blockprofilerate > 0 {
t0 = cputicks()
}
//锁住chan，实现并发安全
lock(&c.lock)
//假如chan现已被封闭，则开释锁，并进行panic
if c.closed != 0 {
unlock(&c.lock)
panic(plainError("send on closed channel"))
}
//假如承受行列中有goroutine，直接将要发送的数据拷贝到接纳的goroutine中
if sg := c.recvq.dequeue(); sg != nil {
send(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true
}
//对于缓冲型的channel，假如还有缓冲空间
if c.qcount < c.dataqsiz {
//知道发送的元素的在buf中的方位
qp := chanbuf(c, c.sendx)
if raceenabled {
racenotify(c, c.sendx, nil)
}
//将数据从ep拷贝到qp
typedmemmove(c.elemtype, qp, ep)
//发送索引加一
c.sendx++
//假如发送索引等于唤醒链表的长度，则将游标贵0
if c.sendx == c.dataqsiz {
c.sendx = 0
}
//缓冲区元素加1，并开释锁
c.qcount++
unlock(&c.lock)
return true
}
//假如不需求堵塞，则直接回来 ，并开释锁
if !block {
unlock(&c.lock)
return false
}
// 获取当时goroutine的指针，
gp := getg()
mysg := acquireSudog()
mysg.releasetime = 0
if t0 != 0 {
mysg.releasetime = -1
}
// No stack splits between assigning elem and enqueuing mysg
// on gp.waiting where copystack can find it.
mysg.elem = ep
mysg.waitlink = nil
mysg.g = gp
mysg.isSelect = false
mysg.c = c
gp.waiting = mysg
gp.param = nil
//当时goroutine进入的发送等候行列
c.sendq.enqueue(mysg)
//告诉其他goroutine，并将goroutine挂起
atomic.Store8(&gp.parkingOnChan, 1)
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
//在没有其他的接纳行列将数据复制到行列中时分，需求确保当时需求被发送的的值一直是可用状况
// sudog 有一个指向仓库目标的指针，可是sudog不是仓库追踪器的根
KeepAlive(ep)
if mysg != gp.waiting {
throw("G waiting list is corrupted")
}
//更新goroutine相关的目标信息
gp.waiting = nil
gp.activeStackChans = false
closed := !mysg.success
gp.param = nil
if mysg.releasetime > 0 {
blockevent(mysg.releasetime-t0, 2)
}
mysg.c = nil
//开释sudog目标
releaseSudog(mysg)
//假如channel现已封闭
if closed {
// close标志位为0，则抛出假性唤醒反常
if c.closed == 0 {
throw("chansend: spurious wakeup")
}
//直接panic
panic(plainError("send on closed channel"))
}
return true
}
发送操作简略流程
1、假如channel是一个空值，在非堵塞形式下，会直接回来。在堵塞形式下，会调用gopark函数挂起goroutine；在非堵塞形式下快速检测到失败并进行回来
2、当channel没准备好接纳时（非缓冲型，等候行列中没有goroutine在等候；缓冲型，buf中没有元素），假如channel现已被封闭，直接回来false；不然，加锁，假如channel现已封闭，而且循环数组buf中没有元素，对应非缓冲型封闭和缓冲型封闭可是buf无元素时，回来对应类型的零值，可是received标志为false，告诉调用者此channel现已封闭，取出来的值并不是正常发送者发送来的数据。
3、假如等候接纳行列recvq不为空，阐明缓冲区中没有数据或许没有缓冲区，此刻直接从recvq取出G,并把数据写入，最终把该G唤醒，完毕发送进程；
4、假如缓冲区中有空余方位，将数据写入缓冲区，完毕发送进程；
5、假如缓冲区中没有空余方位，将待发送数据写入G，将当时G参加sendq，进入睡觉，等候被读goroutine唤醒；
简略发送的流程图
channel发送数据示例
//线程1接纳数据
func goroutineA(vChanA <- chan int) {
val := <- vChanA
fmt.Println("G1 received data :", val)
return
}
//线程2接纳数据
func goroutineB(vChanB <- chan int) {
val := <- vChanA
fmt.Println("G2 received data :", val)
return
}
func main() {
sendChan := make(chan int)
go goroutineA()
go goroutineB()
//发送数据
sendChan <- 1
time.Sleep(time.Second)
}
从channel中接纳数据
channel接纳数据是有chanrecv1和chanrecv2两个函数完结，底层调用的是chanrecv函数。
chanrecv函数源代码
// entry points for <- c from compiled code
//go:nosplit
func chanrecv1(c *hchan, elem unsafe.Pointer) {
chanrecv(c, elem, true)
}
//go:nosplit
func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {
_, received = chanrecv(c, elem, true)
return
}
//chanrecv从channel c中接纳数据，并将接纳到的数据写入到ep中
//当ep为空时，接纳的数据将会被疏忽
//假如goroutine对错堵塞，切元素为空，直接return false
// 不然假如c现已封闭，ep被复制为0，并回来false
// 不然将数据填充导ep指针中，并回来true
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
// raceenabled: don't need to check ep, as it is always on the stack
// or is new memory allocated by reflect.
if debugChan {
print("chanrecv: chan=", c, "\n")
}
if c == nil {
if !block {
return
}
gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
throw("unreachable")
}
// 假如对错堵塞操作，而且c为空
if !block && empty(c) {
//调查到channel没有做好接纳的准备，则调查channel是否现已报备封闭
//原子加载通道假如封闭则回来，因为通道不能从头翻开，后来调查到的通道没有被封闭，就意味着第一次调查的
//时分也没有被封闭，则标明接纳操作不能持续往下，直接return
if atomic.Load(&c.closed) == 0 {
return
}
//通道是不可逆转的封闭，从头检查是否有待接纳的数据，这些数据数据可能是在上面的空和封闭检查之前到达的
// 在这种发送场景下，咱们需求坚持发送的次序共同。
if empty(c) {
// 经过通道是不可逆的且是封闭的
if raceenabled {
raceacquire(c.raceaddr())
}
if ep != nil {
typedmemclr(c.elemtype, ep)
}
return true, false
}
}
var t0 int64
if blockprofilerate > 0 {
t0 = cputicks()
}
//加锁
lock(&c.lock)
//channel现已封闭，而且循环数组buf中没有元素，
//这儿能够处理非缓冲型封闭和缓冲型封闭可是buf无元素的状况
//也便是说即使是封闭状况，但在缓冲型channel，buf有元素的状况下还能接纳元素
if c.closed != 0 && c.qcount == 0 {
if raceenabled {
raceacquire(c.raceaddr())
}
//解锁
unlock(&c.lock)
if ep != nil {
//从一个现已封闭的channel中履行接纳操作，且未疏忽回来值
//那么接纳的值是该类型的一个零值
//typedmemclr 依据类型整理相应地址的内存。
typedmemclr(c.elemtype, ep)
}
return true, false
}
//等候发送的行列有goroutine存在，阐明buf是满的
//这可能是
// 1、非缓冲型的channel
// 2、缓冲型的channel，可是buf现已满了
//针对1，直接进行内存拷贝（从sender goroutine -> receiver goroutine）
//针对2，接纳到循环数组的头部元素，并将发送者的元素放到循环数组尾部
if sg := c.sendq.dequeue(); sg != nil {
recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true, true
}
//缓冲型，buf中有元素，能够正常接纳数据
if c.qcount > 0 {
// 直接从buf中寻觅导对应的元素
qp := chanbuf(c, c.recvx)
if raceenabled {
racenotify(c, c.recvx, nil)
}
if ep != nil {
typedmemmove(c.elemtype, ep, qp)
}
// 整理掉循环数组里相应方位的值
typedmemclr(c.elemtype, qp)
//接纳游标向前移动
c.recvx++
//假如接纳游标 等于环形链表的值，则接纳游标清零。
if c.recvx == c.dataqsiz {
c.recvx = 0
}
//循环数组buf数量减一
c.qcount--
unlock(&c.lock)
return true, true
}
if !block {
// 非堵塞接纳，解锁。selected 回来 false，因为没有接纳到值
unlock(&c.lock)
return false, false
}
//被堵塞的状况，直接结构一个sudog
gp := getg()
mysg := acquireSudog()
mysg.releasetime = 0
if t0 != 0 {
mysg.releasetime = -1
}
// 待接纳的地址被保存下来
mysg.elem = ep
mysg.waitlink = nil
gp.waiting = mysg
mysg.g = gp
mysg.isSelect = false
mysg.c = c
gp.param = nil
//进入channel的等候接纳行列中
c.recvq.enqueue(mysg)
//向任何企图减缩仓库的目标发送信号，标明进程即将停在一个通道上。
atomic.Store8(&gp.parkingOnChan, 1)
//将goroutine挂起
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)
// 被唤醒了，接着从这儿持续履行一些扫尾作业
if mysg != gp.waiting {
throw("G waiting list is corrupted")
}
gp.waiting = nil
gp.activeStackChans = false
if mysg.releasetime > 0 {
blockevent(mysg.releasetime-t0, 2)
}
success := mysg.success
gp.param = nil
mysg.c = nil
releaseSudog(mysg)
return true, success
}
数据读取的简略进程
- 假如等候发送的行列sendq不为空，且没有缓冲区，直接从sendq中取出G，把G中数据读出，最终把G唤醒，完毕读取进程
- 假如等候发送行列sendq不为空，此刻阐明缓冲区已满，从缓冲区首部读出数据，把G中数据写入缓冲区尾部，把G唤醒，完毕读取进程
- 假如缓冲区中有数据，则从缓冲区中取出数据，完毕读取进程
- 将当时goroutine参加recvq，进入睡觉，等候被写goroutine唤醒。
数据读取简略进程的流程图
封闭channel
channel的封闭由closechan函数完结
closechan源码
func closechan(c *hchan) {
//封闭一个nil的channel直接panic
if c == nil {
panic(plainError("close of nil channel"))
}
//加锁
lock(&c.lock)
//假如channel现已被封闭，直接panic
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
var glist gList
// 将 channel 一切等候接纳行列的里 sudog 开释
for {
//接纳行列中出一个sudog
sg := c.recvq.dequeue()
//出队完结，跳出循环
if sg == nil {
break
}
//假如元素不为空，阐明此receiver未疏忽接纳数据
//给它赋一个相应类型的0值
if sg.elem != nil {
typedmemclr(c.elemtype, sg.elem)
sg.elem = nil
}
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
//取出goroutine
gp := sg.g
gp.param = unsafe.Pointer(sg)
sg.success = false
if raceenabled {
raceacquireg(gp, c.raceaddr())
}
//推入导链表中
glist.push(gp)
}
// 将channel中等候接纳行列里的sudog开释
//假如存在这些goroutine将会panic
for {
//从发送行列中出一个sudog
sg := c.sendq.dequeue()
//出队完毕，直接跳出循环
if sg == nil {
break
}
//发送者panic
sg.elem = nil
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
gp := sg.g
gp.param = unsafe.Pointer(sg)
sg.success = false
if raceenabled {
raceacquireg(gp, c.raceaddr())
}
//推入链表中
glist.push(gp)
}
//开释锁
unlock(&c.lock)
//遍历链表
for !glist.empty() {
//取最终一个
gp := glist.pop()
gp.schedlink = 0
//唤醒goroutine
goready(gp, 3)
}
}
封闭channel总结
封闭channel会把recvq中的G悉数唤醒，本该写入G的数据方位为nil。把sendq中的G悉数唤醒，但这些G会panic
除此之外，panic呈现的场景还有
- 封闭值为nil的channel
- 封闭现已被封闭的channel
- 想现已封闭的channel写数据
运用channel的几点不方便的当地
1、在不改变channel本身的状况下，无法获悉一个channel是否封闭
2、封闭一个close channel会导致panic。所以，假如封闭channel的一方在不知道channel是否处于封闭状况是去默然封闭channel是很风险的作业。
3、向一个close的channel发送数据会panic。所以，假如向channel发送数据的一方不知道channel是否处于封闭状况就贸然向channel发送数据是很风险的作业。
封闭channel示例
func IsChanClosed(ch <- chan T) bool {
select {
case <- ch :
return true
default:
}
return false
}
func main() {
c := make(chan T)
//打印得到false
fmt.Println(IsChanClosed(c))
close(c)
//打印得到true
fmt.Println(IsChanClosed(c))
}
本示例仅仅一个粗糙的示例，回来的结果仅仅代表调用的那个瞬间，并不能确保后续调用会不会有其他goroutine改变其状况。
而如何优雅的进行封闭呢 ？答案是只需求添加一个传递封闭信号的channel，receiver经过信号channel下达封闭channel指令。sender监听到封闭信号后，中止发送数据。代码如下
func main() {
rand.Seed(time.Now().UnixNano())
const Max = 10000
const senderNums = 1000
dataCh := make(chan int, 100)
stopChn := make(chan struct{})
//发送数据
for i := 0;i<senderNums;i++ {
go func(){
for {
select {
case <- stopChan:
return
case dataCh <- rand.Intn(Max)
}
}
}
}
//接纳数据
go func() {
for value := range dataCh {
if value == Max - 1 {
fmt.Println("send stop signal to senders.")
close(stopCh)
return
}
fmt.Println(value)
}
}()
//履行等候一个小时
select {
case <- time.After(time.Hour)
}
}
操作channel总结
| 操作 | nil channel | closed channel | 正常的channel |
|---|---|---|---|
| close | panic | panic | 正常封闭 |
| 读 <-ch | 堵塞 | 读对应类型的零值 | 堵塞或许正常读数据。缓冲型为空或非缓冲型没有等候发送者时分为空 |
| 写 ch <- | 堵塞 | panic | 堵塞或许正常写入。非缓冲型没有等候承受者或许缓冲型buf满时会堵塞。 |
channel的运用场景
1、中止信号
channel 用于中止信号的场景仍是挺多的，经常是封闭某个 channel 或许向 channel 发送一个 元素，使得接纳 channel 的那一方获悉道此信息，进而做一些其他的操作。
2、使命守时
与 timer 结合，一般有两种玩法:实现超时操控，实现守时履行某个使命。有时分，需求履行某项操作，但又不想它耗费太长时间，上一个守时器就能够搞定:
select {
case <- time.After(100 * time.Millisecond):
case <- s.stopc :
return fase
}
守时履行某个使命也简略，如下是每隔1s履行使命示例
func worker() {
ticker := time.Tick(1 * time.Second)
for {
select {
case <- ticker :
fmt.Println("履行1s守时使命")
}
}
}
3、解耦出产方和消费方
服务发动时，发动n个worker作为作业协程池，这些协程池作业在一个for循环里，从某个channel消费并履行。如下比如所示，10个协程不断的从作业行列中取使命，出产方只管往channel中发送使命，解耦了出产方和消费方。
func main() {
taskCh := make(chan int, 100)
go worker(taskCh)
//堵塞使命
for i := 0;i<100;i++ {
taskCh <- i
}
//等候一个小时
select {
case <- time.After(time.Hour):
}
}
func worker(taskCh <- chan int) {
const N = 10
for i:=0;i<N;i++ {
go func(index int){
for {
task := <-taskCh
fmt.Println("task : %d is done by worker :%d \n", task, index)
}
}(i)
}
}
