约束
约束可以减轻开发者的认知负担以便写出有更小临界区的并发代码。确保某一信息再并发过程中仅能被其中之一的进程进行访问。程序中通常存在两种可能的约束：特定约束和词法约束。
特定约束
通过公约实现约束，无论是由语言社区、你所在的团队，还是你的代码库设置。在 Go 语言官方默认安装 gofmt 去格式化你的代码，争取让大家都写一样的代码
词法约束
设计使用词法作用域仅公开用于多个并发进程的正确数据和并发原语，这使得做错事是不可能的，例如：Go 中 goroutine 和 channel ，而不是使用 Thread 包（无论是官方，第三方）。在 Go 的世界里操作系统线程不用程序员管理，需要并发 go 就可以了。
for-select 循环
在 Go 语言中你经常看到 for-select 循环。它的结构类似这样的
for{ // 无限循环或者用 range 语句循环
select {
// 使用 channel 的任务
}
}
向 channel 发送数据
for _,v := range []string{"jisdf","jisdf","ier"}{
select {
case <- done:
return
case stringChan <- v:
// 做些什么
}
}
循环等待停止
// 第一种保持 select 语句尽可能短：
// 如果完成的 channel 未关闭，我们将退出 select 语句并继续执行 for 循环
for {
select {
case <- done:
return
default:
}
// 非抢占业务
}
// 第二种将工作嵌入 select 的 default 中
// 如果完成的 channel 尚未关闭，则执行 default 内容的任务
for {
select {
case <- done:
return
default:
// 非抢占业务
}
}
防止 goroutine 泄露
线程安全，是每一个程序员经常讨论的话题。 在 Go 中对应的是 goroutine 协程，虽然 goroutine 开销非常小，非常廉价，但是过多的 goroutine 未得到释放或终止，也是会消耗资源的。goroutine 有以下几种方式被终止：
- 当它完成了它的工作。
- 因为不可恢复的错误，它不能继续工作。
- 当它被告知需要终止工作。
前两种方式非常简单明了，并且隐含在你的程序中。那么我们如何来取消工作？Go 程序在运行时默认会有一个主 goroutine （main goroutine），他会将一些没有工作的 goroutine 设置为自旋，这会导致内存利用率的下降。思考下，既然 main goroutine 能够将其他 goroutine 设置自旋，那么它能不能通知其他 goroutine 停止或退出呢？Of sure ，首先我们需要一个 channel 辅助 main goroutine，它可以包含多种指令，例如超时、异常、特定条件等 。它通常被命名为 done，并且只读。举个例子：
doWork := func(done <- chan int ,s <-chan string) <-chan s{
terminated := make(chan int)
go func () {
// 当前函数 return 后打印一条信息用于验证，for {} 死循环是否被终止
defer fmt.Println("doWork exited")
defer close(termainted)
for {
select {
case l := <- s:
fmt.Println(l)
case <- done: // 由于 select 会相对均匀的挑选 case ，当 done 被读取，则 return 跳出整个并发
return
}
}
}()
return terminated
}
// 创建控制并发的 channel done
done := make(chan int)
terminated := doWork(done, "a")
// 启动一个 goroutine 在 1s 后关闭 done channel
go func() {
time.Sleep(1 * time.Second)
fmt.Println("取消工作的 goroutine")
close(done)
}()
// main goroutine 中读出 termainated 中的数据，验证我们是否成功通知工作的 goroutine 终止工作
<- terminated
fmt.Println("Done")
当一个 goroutine 阻塞了向channel 进行写入的请求，我们可以这样做：
newRandstream := func(done <-chan interface{}) <- chan int{
randStream := make(chan int)
go func(){
defer fmt.Println("newRanstream 关闭了")
defer close(randStream)
for{
select {
case randStream <- rand.int():
case <-done:
return
}
}
}()
return
}
done := make(chan interface{})
randStream := newRandStream(done)
fmt.Println("遍历三次")
for i := 1; i<=3;i++{
fmt.Println("%d: %d\n",i,<-randStream)
}
close(done)
// 模拟正在进行的工作，暂停 1s
time.Sleap(1 * time.Second)
or-channel
以上部分我们了解到单一条件下如何取消 goroutine 防止泄露。如果我们有多种条件触发取消 goroutine ，我们要怎么办呢？让我来了解下 or-channel，创建一个复合 done channel 来处理这种复杂情况。
我们以使用更多的 goroutine 为代价，实现了简洁性。f(x)=x/2 ，其中 x 是 goroutine 的数量，但你要记住 Go 语言种的一个优点就是能够快速创建，调度和运行 goroutine ，并且该语言积极鼓励使用 goroutine 来正确建模问题。不必担心在这里创建的 goroutine 的数量可能是一个不成熟的优化。此外，如果在编译时你不知道你正在使用多少个 done channel ，则将会没有其他方式可以合并 done channel。
错误处理
说到错误处理，也许很多程序程序员觉得 Go 语言错误处理简直太糟糕了。漫天的 if err != nil{} ，try catch 捕捉并打印错误多么好。我要说首先我们需要注意 Go 的并发模式，与其他语言有着很大的区别。Go 项目开发者希望我们将错误视为一等公民，合并入我们定义的消息体内，channel 中的数据被读出的时候我们进行判断，程序并发过程中是否出现错误。这避免了多进程多线程模型下，try catch 丢失一些报错，在故障回顾的时候非常麻烦。
// 建议的消息体
type MyMessage struct{
Data string
Err error
}
让错误成为一等公民合并进你的结构体中，代码也许会更易懂
type MyMessage struct{
N int
Err error
}
func myfuncation(n string) MyMessage{
var mm MyMessage
mm.N,mm.Err = anotherFunc(n)
return mm
}
func anotherFunc(n string) (int,error){
i,err := strconv.Atoi(n)
if err !=nil{
return i,err
}
return i,nil
}
func main(){
mymsg := myfuncation("Concurrency In GO")
if mymsg.Err != nil{
// 这里可以换成其他的 log 框架，部分 log 框架会自动识别 error 来源。例如：func (m *MyMessage) myfuncation() 这样的函数就会被抓到错误来自于哪里。
fmt.Println(mymsg.Err)
}
}
pipeline
我曾经在祖传代码中见到一个约 2000 行的函数。我希望看见这篇文章的你，不要这么做。我们已经了解了数据如何在两个或多个 goroutine 之间通过 channel 传递，那我我们把这样的程序用多个 channel组合在一起，其中的每一次读出，或写入channel 都是这一环上的一个 stage（步），这就是 pipeline。Go 语言的并发模式，让我们很方便，快捷，安全的在一个进程中实现了流式处理。我们来看一个官方 pipeline 的例子：
package main
import (
"fmt"
"sync"
"time"
)
func gen(nums ...int) <-chan int {
genOut := make(chan int)
go func() {
for _, n := range nums {
genOut <- n
}
fmt.Println("Input gen Channel number =>", len(genOut))
close(genOut)
}()
return genOut
}
func sq(done <-chan struct{}, in <-chan int) <-chan int {
sqOut := make(chan int)
go func() {
// 这个 close(sqOut) 一定要先写，执行的时候优先压入栈，待函数执行完成关闭 sqOut channel
defer close(sqOut)
for n := range in {
// 利用 select {} 均衡调度 channel
select {
case sqOut <- n * n:
fmt.Printf("=> %v <= write into sqOut channel \n", n*n)
case <-done:
return
}
}
//fmt.Printf("Wait close the chan => %v\n", len(sqOut))
}()
return sqOut
}
// merge Fan-In 函数合并多个结果
func merge(done <-chan struct{}, cs ...<-chan int) <-chan int {
var wg sync.WaitGroup
mergeOut := make(chan int, 1)
output := func(c <-chan int) {
defer wg.Done()
for n := range c {
select {
case mergeOut <- n:
case <-done:
return
}
}
}
wg.Add(len(cs))
for _, c := range cs {
go output(c)
}
go func() {
wg.Wait()
close(mergeOut)
}()
return mergeOut
}
// pfnumber 计算算数平方数
func pfnumber() {
// 定义 don channel 用于终止 pipeline
don := make(chan struct{}, 3)
don <- struct{}{}
don <- struct{}{}
close(don)
// 传入 don 通知发送方停止发送
for n := range sq(don, sq(don, gen(3, 4, 2))) {
fmt.Println("Last result ", n)
}
fmt.Println("============================================")
}
func fanInOut() {
don := make(chan struct{}, 3)
in := gen(2, 3)
c1 := sq(don, in)
c2 := sq(don, in)
for n := range merge(don, c1, c2) {
fmt.Println(n)
}
don <- struct{}{}
don <- struct{}{}
don <- struct{}{}
fmt.Println("Finish channel len => ", len(don))
<-don
close(don)
}
func f1(i chan int) {
fmt.Println(<-i)
}
func runf1() {
out := make(chan int)
go f1(out)
time.Sleep(2 * time.Second)
out <- 2
time.Sleep(2 * time.Second)
}
func main() {
//runf1()
pfnumber()
// FanIn and FanOut
//fanInOut()
}
简单总结一下如何正确构建一个 pipeline:
- 当所有的发送已完成，stage 应该关闭输出 channel
- stage 应该持续从只读 channel 中读出数据，除非 channel 关闭或主动通知到发送方停止发送
扇出、扇入
扇出模式优先的场景：
- 它不依赖于之前的 stage 计算的值
- 需要运行很长时间，例如：I/O 等待，远程调用，访问 REST full API等
扇入模式优先：
扇入意味着多个数据流复用或者合并成一个流。例如：上文 pipeline 中的 merge 函数，可以通过打开 fanInOut() 函数执行一下试试。
or-done-channel
在防止 goroutine 泄露，pipeline 中我们都在函数执行过程中嵌入了 done channel 以便终止需要停止的 goroutine。我们可以看出他们有个统一的特点，传入 done ,jobChannel ，返回 resultChannel 。那么我们可以把它封装起来，像这样：
orDone := func(done ,c <-chan interface{}) <- chan interface{}{
valStream := make(chan interface{})
go func(){
defer close(valStream)
for {
select{
case <- done:
case v,ok := <- c:
if ok == false{
return
}
select{
case valStream <- v:
case <-done:
}
}
}
}()
return valStream
}
tee-channel
可能需要将同一个结果发送给两个接收者，这个时候就需要用到 tee-channel 的方式。
应用场景：
- 流量镜像
- 操作审计
tee := func(done <- chan interface{},in <-chan interface{}
)(_,_ <- chan interface{}) { <-chan interface{}) {
out1 := make(chan interface{})
out2 := make(chan interface{})
go func(){
defer close(out1)
defer close(out2)
for val := range orDone(done, in){
var out1,out2 = out1,out2
for i:=0;i<2; i++{
select{
case <- done:
case out1 <- val:
out1 = nil
case out2 <- val:
out2 = nil
}
}
}
}()
return out1,out2
}
其他的应用场景
桥接 channel
在 channel 中传递 channel 。笔者学术才浅，纸上谈兵多，动手实践少，着实想不到合适的场景，希望读者能为我补充一下。
队列
队列可能是我们第一次看见 channel 的感受，这玩意一个队列，非常具备队列的特性。
队列在什么样的情况下可以提升整体性能
- 如果在一个 stage 批处理请求可以节省时间。
- 需要缓存的场景，例如：批量日志刷盘，热数据缓存等。
context 包
在前文中经常会定义 done channel 的做法，防止 goroutine 泄露，或者主动中断需要停止的 pipeline 。难道我们每次构建 pipeline 的时候都要创建 done channel 吗？答案是否定的，Go 团队为我们准备了 context 包，专用于干类似的工作。
type Context interface {
// 当该 context 工作的 work 被取消时，返回超时时间
Deadline() (deadline time.Time, ok bool)
// done 返回停止 pipeline 的 channel
Done() <chan struct{}
// error 一等公民。
// 如果 context 被取消，超时，返回取消，超时的原因，以 error 形式返回。
Err() error
// 返回与此 context 关联的 key
Value(key interface{}) interface{}
}
context 包有两个主要目的：
- 提供一个可以取消你的调用意图中分支的 API.
- 提供用于通过呼叫传输请求范围数据的数据包
在 防止 goroutine 泄露中学到，函数中的取消有三个方面，context 包可以帮你管理它：
- goroutine 的父 goroutine 可能想要取消它。
- 一个 goroutine 可能想要取消它的子 goroutine。
- goroutine 中任何阻塞操作都必须是可抢占的 ，以便它可以被取消。
Context.Value(key interface{}) ，由于使用 interface{} 作为函数参数，这里我们需要强调一下使用注意事项，及建议：
- 虽然可以在 context.Context 中出传递 value，但是并不建议这么做，因为我们需要保证这个值必须是安全的，可以被多个 goroutine 访问。要知道不通的 goroutine 处理逻辑可能是不同的。
- 值传递适合在远程 API 调用时使用，请勿在进程内使用。
- 数据应该时不可变的。
- 使用简单类型，例如：int,float,string 等基础类型。
- 数据应该是数据，而不是类型与方法。
- 数据应该用于修饰操作，而不是驱动操作
结束语
第四章可以称之为全书核心章节，它将前面的部分总结归纳，并形成很多的 Go 语言并发技巧讲解，可以帮助我们写出可维护的并发代码。熟悉了这些并发模式，我们可以将多种模式组合，以帮助我们编写大型系统。
笔者能力优先，才疏学浅，希望读者能够翻阅原书，深入理解并充分运用在工作中。