Go（5）--并发

迷南。 2021-08-29 17:18 380阅读 0赞

### 并发 ###

*  1. 并发基础
 *   *  1.1 并发和并行
     *  1.2 goroutine
     *  1.3 chan
     *  1.4 WaitGroup
     *  1.5 select
     *  1.6 扇入和扇出
     *  1.7 通知退出机制
 *  2. 并发范式
 *   *  2.1 生成器
     *  2.2 管道
     *  2.3 每个请求一个 goroutine
     *  2.4 固定 worker 工作池
     *  2.5 future 模式
 *  3. context 标准库
 *   *  3.1 context 的设计目的
     *  3.2 基本数据结构
     *   *  3.2.1 Context 接口
         *  3.2.2 canceler 接口
         *  3.2.3 empty Context 结构
         *  3.2.4 cancelCtx
         *  3.2.5 timerCtx
         *  3.2.6 valueCtx
     *  3.3 API 函数
     *  3.4 辅助函数
     *  3.5 context 的用法
 *  4. 并发模型
 *   *  4.1 CSP 简介
     *  4.2 调度模型
     *  4.3 并发和调度
     *   *  4.3.1 G（Goroutine）
         *  4.3.2 M（Machine）
         *  4.3.3 P（Processor）

# 1. 并发基础 #

## 1.1 并发和并行 ##

*  并行：程序在任意时刻都是同时运行的；比如说：在吃饭的时候，可以听音乐。不用停下吃饭，就可以听音乐。
 *  并发：程序在单位时间内是同时运行的；比如说：在吃饭的时候喝水，需要停下吃饭，然后去喝水。

## 1.2 goroutine ##

>       操作系统可以进行线程和进程的调度，本身具备并发处理能力，但进程切换代价还是过于高昂，进程切换需要保存上下文，耗费较多的时间。而 Go 的并发基础便是基于应用程序在用户层再构筑一级调度，将并发的粒度进一步降低，避免频繁从用户态到内核态切换。

Go 语言的并发执行体称为`goroutine`，通过`go`关键字来启动一个`goroutine`。注意：`go`关键字后面必须跟一个函数，不能是语句或其他东西，函数的返回值被忽略。

*  通过 go + 匿名函数形式启动`goroutine`，代码如下：

package main
    
    import (
    	"fmt"
    	"runtime"
    	"time"
    )
    
    func main() { 
    	go func() { 
    		sum := 0
    		for i := 0; i < 10000; i++{ 
    			sum += i
    		}
    		fmt.Println(sum)
    		time.Sleep(1 * time.Second)
    	}()
    
    	// 返回当前程序的 goroutine 数目
    	fmt.Println("NumGoroutine = ", runtime.NumGoroutine())
    
    	// main goroutine sleep 5 秒，防止其提前退出
    	time.Sleep(5 * time.Second)
    }

![在这里插入图片描述][20200517094109609.png]

*  通过 go + 有名函数形式启动`goroutine`，代码如下：

package main
    
    import (
    	"fmt"
    	"runtime"
    	"time"
    )
    
    func sum() { 
    	sum := 0
    	for i := 0; i < 10000; i++{ 
    		sum += i
    	}
    	fmt.Println(sum)
    	time.Sleep(1 * time.Second)
    }
    
    func main() { 
    	go sum()
    
    	// 返回当前程序的 goroutine 数目
    	fmt.Println("NumGoroutine = ", runtime.NumGoroutine())
    
    	// main goroutine sleep 5 秒，防止其提前退出
    	time.Sleep(5 * time.Second)
    }

![在这里插入图片描述][20200517094201763.png]

**`goroutine` 有如下特性**：

*  go 的执行是非阻塞的，不会等待。
 *  go 后面的函数的返回值会被忽略。
 *  调度器不能保证多个 `goroutine`的执行次序。
 *  没有父子`goroutine`的概念，所有的`goroutine`是平等地调度和执行地。
 *  Go 程序在执行时会单独为 `main`函数创建一个`goroutine`，遇到其他`go`关键字时再去创建其他的`goroutine`。
 *  Go 没有暴露 `goroutine id`给用户，所以不能在一个`goroutine`里面显式地操作另一个`goroutine`，不过`runtime`包提供了一些函数访问和设置`goroutine`的相关信息。

1、`func GOMAXPROCS`

`func GOMAXPROCS(n int) n`用来设置或查询可以并发执行的`goroutine`数目，n 大于1表示`GOMAXPROCS`值，否则表示查询当前的`GOMAXPROCS`值。

package main
    
    import "runtime"
    
    func main() { 
    	// 获取当前的 GOMAXPROCS 值
    	println("GOMAXPROCS = ", runtime.GOMAXPROCS(0))
    
    	// 设置 GOMAXPROCS 的值为2
    	runtime.GOMAXPROCS(2)
    
    	// 获取当前的 GOMAXPROCS 值
    	println("GOMAXPROCS = ", runtime.GOMAXPROCS(0))
    }

![在这里插入图片描述][2020051709561626.png]

2、`func Goexit`

`func Goexit()`是结束当前`goroutine`的运行，`Goexit`在结束当前`goroutine`运行之前会调用当前`goroutine`已经注册的`defer`。`Goexit`并不会产生`panic`，所以该`goroutine defer`里面的 `recover`调用都返回`nil`。

3、`func Gosched`

`func Gosched()`是放弃当前调度执行机会，将当前`goroutine`放到队列中等待下次被调度。

## 1.3 chan ##

`chan`是 Go 语言里面的一个关键字，是`channel`的简写，即通道。通道是 `goroutine`之间通信和同步的重要组件，Go的哲学是“**不要通过共享内存来通信，而是通过通信来共享内存**”，通道是 Go 通过通信来共享内存的载体。

**channel的介绍**  
            1、channel 本质就是一个数据结构-队列  
            2、数据是先进先出  
            3、**线程安全**，多 goroutine 访问时，不需要加锁，就是说 channel 本身就是线程安全的。  
            4、channel 是有类型的，一个 string 的channel 只能存放 stringg 类型数据。

通道是有类型的，可以简单地把它理解为有类型地管道。声明一个简单地通道语句是`chan dataType`，但是简单声明一个通道变量没有任何意义，并没有被初始化，为`nil`。Go 提供一个内置函数`make`来创建通道。

// 创建一个无缓冲的通道，通道存放元素的类型为 dataType
    make(chan dataType)
    
    // 创建一个有 10个缓冲的通道，通道存放元素的类型为 dataType
    make(chan dataType, 10)

通道分为无缓冲通道和有缓冲通道，Go 提供内置函数`len`和`cap`，无缓冲的通道的`len`和`cap`都是0，有缓冲的通道的`len`代表没有被读取的元素数，`cap`代表整个通道的容量。无缓冲的通道既可以用于通信，也可以用于两个`goroutine`的同步，有缓冲的通道主要用于通信。

在上例中，我们为了避免`main goroutine`过早退出，所以`sleep`了5秒，但是这个时间并不合理。如果另一个`goroutine`执行时间大于5秒，那么还未执行完便已经结束了。而如果执行时间不用5秒，显然5秒的时间依然浪费了CPU资源。而有了通道之后，便可以使用无缓冲的通道来实现`goroutine`之间的同步等待。

package main
    
    import (
    	"runtime"
    )
    
    func sum(c chan struct{ }) { 
    	sum := 0
    	for i := 0; i < 10000; i++{ 
    		sum += i
    	}
    	println(sum)
    	// goroutine 执行完后
    	// 写通道
    	c <- struct{ }{ }
    }
    
    func main() { 
    	c := make(chan struct{ })
    
    	go sum(c)
    
    	// 返回当前程序的 goroutine 数目
    	println("NumGoroutine = ", runtime.NumGoroutine())
    
    	// 读通道c，通过通道进行同步等待
    	<- c
    }

`goroutine`运行结束后退出，写道缓冲通道中的数据不会消失，它可以缓冲和适配两个`goroutine`处理速率不一致的情况。

package main
    
    import "runtime"
    
    func main() { 
    	var c chan struct{ }
    	var ci chan int
    
    	c = make(chan struct{ })
    	ci = make(chan int, 100)
    
    	go func(i chan struct{ }, j chan int) { 
    		for i := 0; i < 10; i++{ 
    			ci <- i
    		}
    		close(ci)
    		c <- struct{ }{ }
    	}(c, ci)
    
    	println("NumGoroutine = ", runtime.NumGoroutine())
    
    	// 读通道 c，通过通道进行同步等待
    	<-c
    
    	// 此时 ci 通道已经关闭，匿名函数启动的 goroutine 已经退出
    	println("NumGoroutine = ", runtime.NumGoroutine())
    
    	// 但通道 ci 还可以继续读取
    	for v := range ci{ 
    		println(v)
    	}
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70]  
      操作不同状态的`chan`会引发三种行为：

*  **panic**

（1）向已经关闭的通道写数据会导致`panic`。

（2）重复关闭的通道会导致`panic`。

*  **阻塞**

（1）向未初始化的通道写数据或读数据都会导致当前`goroutine`的永久阻塞。

（2）向缓冲区已满的通道写数据会导致`goroutine`阻塞。

（3）通道中没有数据，读取该通道会导致`goroutine`阻塞。

*  **非阻塞**

（1）读取已经关闭的通道不会引发阻塞，而是立即返回通道元素类型的零值，可以使用`comma , ok`语法判断通道是否已经关闭。

package main
    
    import "fmt"
    
    // WriteData
    func WriteData(intChan chan int){ 
        for i := 1; i<= 50; i++{ 
            intChan <- i
        }
        close(intChan)
    }
    
    func ReadData(intChan chan int, exitChan chan bool){ 
        for{ 
            v, ok := <- intChan
            if !ok{ 
                break
            }
            fmt.Println("v = ", v)
        }
        // readData 读取完毕后，任务完成
        exitChan <- true
        close(exitChan)
    }
    
    func main()  { 
        // 创建两个管道
        intChan := make(chan int, 50)
        exitChan := make(chan bool, 1)
        
        go WriteData(intChan)
        go ReadData(intChan, exitChan)
        for{ 
        	// ok 判断 chan 是否被关闭
            _, ok := <- exitChan
            if !ok{ 
                break
            }
        }
    }

（2）向有缓冲且没有满的通道读/写不会引发阻塞。

**`chan`使用细节和注意事项**：

1.  `channel`的遍历

`channel` 支持 `for - range` 的方式进行遍历，注意两个细节：

1、在遍历时，如果 `channel` 没有关闭，则会出现 `deadlock` 错误  
            2、在遍历时，如果 `channel` 已经关闭，则会正常遍历数据，遍历完后，就会退出遍历.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 1]

1.  `channel`可以声明为只读、或者只写性质

![在这里插入图片描述][20200517105437817.png]  
![在这里插入图片描述][20200517105448890.png]

1.  使用 `select` 可以解决从管道取数据的阻塞问题

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 3]  
4. `goroutine` 中使用 `recover`，解决协程中出现 `panic`，导致程序崩溃问题.

说明：如果我们启一个协程，但是这个协程出现了 `panic`，如果我们没有捕获这个 `panic`，就会造成整个程序崩溃，这时我们可以在 `goroutine` 中使用 `recover` 来捕获 `panic`，进行处理，这样即使这个协程发生问题，但是主线程依然不受影响，可以继续执行。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 4]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 5]

## 1.4 WaitGroup ##

`goroutine`用于并发，`chan`用于通信。没有缓冲的通道具有同步的功能，除此之外，`sync`包也提供了多个`goroutine`同步机制，主要通过`WaitGroup`实现。

主要数据结构和操作如下：

type WaitGroup struct { 
    	...	
    }
    
    // 添加等待信号
    func (wg *WaitGroup) Add(delta int){ 
    	
    }
    
    // 释放等待信号
    func (wg *WaitGroup) Done(){ 
    
    }
    
    // 等待
    func (wg *WaitGroup) Wait(){ 
    
    }

`WaitGroup`用来等待多个`goroutine`完成，`main groutine`调用`Add`设置需要等待`goroutine`的数目，每一个`goroutine`结束时调用`Done()`，`Wait()`被`main`用来等待所有的`goroutine`完成。

package main
    
    import (
    	"fmt"
    	"net/http"
    	"sync"
    )
    
    var wg sync.WaitGroup
    
    var urls = []string{ 
    	"http://www.qq.org",
    	"http://www.baidu.com",
    }
    
    func main() { 
    	for _, url := range urls{ 
    		// 每一个 URL 启动一个 goroutine,同时给 wg 加1
    		wg.Add(1)
    
    		go func(url string) { 
    			// 当前 goroutine 结束后给 wg 计数减1，wg.Done() 等价于 wg.Add(-1)
    			defer wg.Done()
    
    			// 发送 HTTP get 请求，并打印 HTTP 返回码
    			resp, err := http.Get(url)
    			if err == nil{ 
    				fmt.Println(resp.Status)
    			}
    		}(url)
    	}
    
    	// 等待所有请求结束
    	wg.Wait()
    }

![在这里插入图片描述][20200517115511911.png]

## 1.5 select ##

`select`是类`UNIX`系统提供的一个多路复用系统`API`，Go语言借用多路复用的概念，提供了`select`关键字，用于多路监听多个通道。**当监听的通道没有状态是可读或可写的，`select`是阻塞的；只要监听的通道中有一个状态是可读或可写的，则`select`就不会阻塞，而是进入处理就绪通道的分支流程**。如果监听的通道有多个可读或可写的状态，则`selct`随机选取一个操作。

package main
    
    func main() { 
    	ch := make(chan int, 1)
    
    	go func(chan int) { 
    		for{ 
    			select { 
    				case ch <- 0:
    				case ch <- 1:
    			}
    		}
    	}(ch)
    
    	for i := 0; i < 10; i++{ 
    		println(<-ch)
    	}
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 6]

## 1.6 扇入和扇出 ##

扇入是指将多路通道聚合到一条通道中处理，Go 语言最简单的扇入就是使用 `select`聚合多条通道服务。

扇出是指将一条通道发散到多条通道中处理，在 Go 语言里面具体实现就是使用 `go`关键字启动多个`goroutine`并发处理。

## 1.7 通知退出机制 ##

读取已经关闭的通道不会引起阻塞，也不会导致`panic`，而是立即返回该通道类型的零值。**关闭`select`监听的某个通道能使`select`立即感知这种通知，然后进行相应的处理**，这就是退出通知机制。

下游的消费者不需要随机数时，显式地通知生产者停止生产。

package main
    
    import (
    	"fmt"
    	"math/rand"
    	"runtime"
    )
    
    // GenerateIntA 是一个随机数生成器
    func GenerateIntA(done chan struct{ }) chan int{ 
    	ch := make(chan int)
    	go func() { 
    		Label:
    			for{ 
    				select { 
    					case ch <- rand.Int():
    
    					// 增加一路监听，就是对退出通知信号 done 地监听
    					case <- done:
    						break Label
    				}
    			}
    		// 收到通知后关闭通道 ch
    		close(ch)
    	}()
    	return ch
    }
    
    func main() { 
    
    	done := make(chan struct{ })
    	ch := GenerateIntA(done)
    
    	fmt.Println(<-ch)
    	fmt.Println(<-ch)
    
    	// 发送通知，告诉生产者停止生产
    	close(done)
    
    	fmt.Println(<-ch)
    	fmt.Println(<-ch)
    
    	// 此时生产者已经退出
    	println("NumGoroutine = ", runtime.NumGoroutine())
    }

![在这里插入图片描述][20200517194911265.png]

# 2. 并发范式 #

## 2.1 生成器 ##

有如下一个应用常营，就是调用一个统一地全局的生成器服务，用于生成全局事务号、订单号、序列号和随机数等。

1.  最简单的带缓冲的生成器

1.  多个`goroutine`增强型生成器

package main
    
    import (
    	"fmt"
    	"math/rand"
    )
    
    func GeneratorIntA() chan int  { 
    	ch := make(chan int, 10)
    
    	// 启动一个 goroutine 用于生成一个随机数，函数返回一个通道用于获取随机数
    	go func() { 
    		for  { 
    			ch <- rand.Int()
    		}
    	}()
    
    	return ch
    }
    
    func GeneratorIntB() chan int  { 
    	ch := make(chan int, 10)
    
    	// 启动一个 goroutine 用于生成一个随机数，函数返回一个通道用于获取随机数
    	go func() { 
    		for  { 
    			ch <- rand.Int()
    		}
    	}()
    
    	return ch
    }
    
    func GeneratorInt() chan int  { 
    	ch := make(chan int, 20)
    
    	// 使用 select 的扇入计数增加生成的随机源
    	go func() { 
    		for  { 
    			select { 
    				case ch <- <- GeneratorIntA():
    				case ch <- <- GeneratorIntB():
    			}
    		}
    	}()
    
    	return ch
    }
    
    
    
    func main() { 
    	ch := GeneratorInt()
    	fmt.Println(<- ch)
    	fmt.Println(<- ch)
    }

1.  有时希望生成器能够自动退出，可以借助`Go`通道的退出通知机制实现。

package main
    
    import (
    	"fmt"
    	"math/rand"
    )
    
    func GeneratorIntA(done chan struct{ }) chan int  { 
    	ch := make(chan int, 10)
    
    	go func() { 
    		label:
    		for  { 
    			// 通过 select 监听一个信号 chan 来确定是否停止生成
    			select { 
    				case ch <- rand.Int():
    
    				case <- done:
    					break label
    			}
    		}
    		close(ch)
    	}()
    
    	return ch
    }
    
    func main() { 
    	done := make(chan struct{ })
    	ch := GeneratorIntA(done)
    	fmt.Println(<- ch)
    	fmt.Println(<- ch)
    
    	// 不再需要生成器，通过 close chan 发送一个通知给生成器
    	close(done)
    
    	for v := range ch{ 
    		println(v)
    	}
    }

1.  一个融合了并发、缓冲、退出通知等多重特性的生成器。

package main
    
    import (
    	"math/rand"
    )
    
    func GeneratorIntA(done chan struct{ }) chan int  { 
    	ch := make(chan int, 10)
    
    	go func() { 
    		label:
    		for  { 
    			// 通过 select 监听一个信号 chan 来确定是否停止生成
    			select { 
    				case ch <- rand.Int():
    
    				case <- done:
    					break label
    			}
    		}
    		close(ch)
    	}()
    
    	return ch
    }
    
    func GeneratorIntB(done chan struct{ }) chan int  { 
    	ch := make(chan int, 10)
    
    	go func() { 
    	label:
    		for  { 
    			// 通过 select 监听一个信号 chan 来确定是否停止生成
    			select { 
    			case ch <- rand.Int():
    
    			case <- done:
    				break label
    			}
    		}
    		close(ch)
    	}()
    
    	return ch
    }
    
    // 通过 select 执行扇入操作
    func GeneratorInt(done chan struct{ }) chan int  { 
    	ch := make(chan int, 20)
    	// 使用 select 的扇入计数增加生成的随机源
    	go func() { 
    		Label:
    		for  { 
    			select { 
    				case ch <- <- GeneratorIntA(done):
    				case ch <- <- GeneratorIntB(done):
    			    case <- done:
    			    	break Label
    			}
    		}
    		close(ch)
    	}()
    	return ch
    }
    
    
    
    func main() { 
    	// 创建一个作为接收退出信号的 chan
    	done := make(chan struct{ })
    	
    	// 启动生成器
    	ch := GeneratorInt(done)
    
    	// 获取生成器资源
    	for i := 0; i < 10; i++{ 
    		println(<- ch)
    	}
    
    
    	// 通知生成者停止生产
    	done <- struct{ }{ }
    	println("stop generator")
    }

## 2.2 管道 ##

通道可以分为两个方向，一个是读，另一个是写，假如一个函数的输入参数和输出参数都是相同的`chan`类型，则该函数可以调用自己，最终形成一个调用链。多个具有相同参数类型的函数也能组成一个调用链，像 UNIX 系统的管道，是一个由类型的管道。

package main
    
    import "fmt"
    
    // chain 函数的输入参数和输出参数类型相同，都是 chan int 类型
    // chain 函数的功能是将 chan 内的数据统一加1
    func chain(in chan int) chan int{ 
    	out := make(chan int)
    	go func() { 
    		for v := range in{ 
    			out <- v + 1
    		}
    		close(out)
    	}()
    	return out
    }
    
    func main() { 
    	in := make(chan int)
    
    	// 初始化输入参数
    	go func() { 
    		for i := 0; i < 10; i++{ 
    			in <- i
    		}
    		close(in)
    	}()
    
    	// 连续调用 3 次 chan，相当于 in 中的每个元素都加3
    	out := chain(chain(chain(in)))
    	for v := range out{ 
    		fmt.Println(v)
    	}
    }

## 2.3 每个请求一个 goroutine ##

为每一个请求或任务创建一个 `goroutine`去处理，典序场景如 Go 中的 Http Server 服务。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 7]

计算 100 个自然数的和，将计算任务拆分为多个`task`，每个`task`启动一个`goroutine`进行处理。

package main
    
    import (
    	"fmt"
    	"sync"
    )
    
    // 工作任务
    type task struct { 
    	begin int
    	end int
    	result chan <- int
    }
    
    // 任务执行，计算 begin 到 end 的和
    // 执行结果写入结果 chan result
    func (t *task)do()  { 
    	sum := 0
    	for i := t.begin; i <= t.end; i++{ 
    		sum += i
    	}
    	t.result <- sum
    }
    
    func main() { 
    	// 创建任务通道
    	taskchan := make(chan task, 10)
    
    	// 创建结果通道
    	resultchan := make(chan int, 10)
    
    	// wait 用于同步等待任务的执行
    	wait := &sync.WaitGroup{ }
    
    	// 初始化 task 的 goroutine，计算 100 个自然数之和
    	go InitTask(taskchan, resultchan, 100)
    
    	// 每个 task 启动一个 goroutine 进行处理
    	go DistributeTask(taskchan, wait, resultchan)
    
    	// 通过结果通道获取结果并汇总
    	sum := ProcessResult(resultchan)
    	fmt.Println("sum = ", sum)
    }
    
    // 创建 task 并写入 task 通道
    func InitTask(taskchan chan <- task, r chan int, p int)  { 
    	qu := p / 10
    	mod := p % 10
    	high := qu * 10
    	for j := 0; j < 10; j++{ 
    		b := j * 10 + 1
    		e := 10 * (j + 1)
    		tsk := task{ 
    			begin: b,
    			end: e,
    			result: r,
    		}
    		taskchan <- tsk
    	}
    	
    	if mod != 0{ 
    		tsk := task{ 
    			begin: high + 1,
    			end: p,
    			result: r,
    		}
    		taskchan <- tsk
    	}
    	close(taskchan)
    }
    
    // 读取 task chan，每个 task 启动一个 worker goroutine 进行处理
    // 并等待每个 task 运行完，关闭结果通道
    func DistributeTask(taskchan <- chan task, wait *sync.WaitGroup, result chan int)  { 
    	for v := range taskchan{ 
    		wait.Add(1)
    		go ProcessTask(v, wait)
    	}
    	wait.Wait()
    	close(result)
    }
    
    // goroutine 处理具体工作，并将处理结果发送到结果通道
    func ProcessTask(t task, wait *sync.WaitGroup)  { 
    	t.do()
    	wait.Done()
    }
    
    // 读取结果通道，汇总结果
    func ProcessResult(result chan int)int  { 
    	sum := 0
    	for v := range result{ 
    		sum += v
    	}
    	return sum
    }

程序逻辑分析：

（1）InitTask 函数创建 task 并发送到 task 通道。

（2）分发任务函数 DistributeTask 为每个 task 启动一个 goroutine 处理任务，等待其处理完成，然后关闭结果通道。

（3）ProcessResult 函数读取并统计所有的结果。

## 2.4 固定 worker 工作池 ##

在 Java 中，我们通过线程池避免频繁创建、销毁线程浪费的资源。在 Go 中，一个可以构建固定数目的 `goroutine`作为工作线程池。

在上述程序中，处理主要的 `main goroutine`，还开启了以下几类 `goroutine`:

1.  初始化任务的`goroutine`
2.  分发任务的`goroutine`
3.  等待所有`worker`结束通知，然后关闭结果通道的`goroutine`

`main`函数负责拉起上述`goroutine`，并从结果通道获取最终的结果。

程序采用3个通道，分别是：

1.  传递 `task` 任务的通道
2.  传递 `task` 结果的通道
3.  接受 `worker` 处理完任务后所发送通知的通道

package main
    
    import "fmt"
    
    // 工作池的 goroutine 数目
    
    const (
    	NUMBER = 10
    )
    
    // 工作任务
    type task struct { 
    	begin int
    	end int
    	result chan <- int
    }
    
    // 任务处理：计算 begin 到 end 的和
    // 执行结果写入结果 chan result
    func (t *task)do()  { 
    	sum := 0
    	for i := t.begin; i <= t.end; i++{ 
    		sum += i
    	}
    	t.result <- sum
    }
    
    func main() { 
    	workers := NUMBER
    
    	// 工作通道
    	taskchan := make(chan task, 10)
    	// 结果通道
    	resultchan := make(chan int, 10)
    
    	// worker 信号通道
    	done := make(chan struct{ }, 10)
    
    	// 初始化 task 的 goroutine，计算 100 个自然数之和
    	go InitTask(taskchan, resultchan, 100)
    
    	// 分发任务到NUMBER个 goroutine 池
    	DistributeTask(taskchan, workers, done)
    
    	// 获取各个 goroutine 处理完任务的通知，并关闭结果通道
    	go CloseResult(done, resultchan, workers)
    
    	// 通过结果通道获取结果并汇总
    	sum := ProcessResult(resultchan)
    
    	fmt.Println("sum = ", sum)
    }
    
    // 创建 task 并写入 task 通道
    func InitTask(taskchan chan <- task, r chan int, p int)  { 
    	qu := p / 10
    	mod := p % 10
    	high := qu * 10
    	for j := 0; j < 10; j++{ 
    		b := j * 10 + 1
    		e := 10 * (j + 1)
    		tsk := task{ 
    			begin: b,
    			end: e,
    			result: r,
    		}
    		taskchan <- tsk
    	}
    
    	if mod != 0{ 
    		tsk := task{ 
    			begin: high + 1,
    			end: p,
    			result: r,
    		}
    		taskchan <- tsk
    	}
    	close(taskchan)
    }
    
    // 读取 task chan 并分发到 worker goroutine 处理，总的数量是 workers
    func DistributeTask(taskchan <- chan task, workers int, done chan struct{ })  { 
    
    	for i := 0; i < workers; i++{ 
    		go ProcessTask(taskchan, done)
    	}
    
    }
    
    // 工作 goroutine 处理具体工作，并将处理结果发送到结果 chan
    func ProcessTask(taskchan <-chan task, done chan struct{ })  { 
    	for t := range taskchan{ 
    		t.do()
    	}
    	done <- struct{ }{ }
    }
    
    // 通过 done chan 同步等待所有 goroutine 的结束，然后关闭 chan
    func CloseResult(done chan struct{ }, resultchan chan int, workers int)  { 
    	for i := 0; i < workers; i++{ 
    		<-done
    	}
    	close(done)
    	close(resultchan)
    }
    
    // 读取结果通道，汇总结果
    func ProcessResult(result chan int)int  { 
    	sum := 0
    	for v := range result{ 
    		sum += v
    	}
    	return sum
    }

程序逻辑分析：

（1）构建 task 并发送到 task 通道。

（2）分别启动 n 个工作线程，不停地从 task 通道中获取任务，然后将结果写入结果通道。如果任务通道被关闭，则负责向收敛结果的`goroutine`发送通知，告诉其当前`worker`已经完成工作。

（3）收敛结果的`goroutine`接收到所有的`task`已经处理完毕的信号后，主动关闭结果通道。

（4）main 函数中的 ProcessResult 函数读取并统计所有的结果。

## 2.5 future 模式 ##

编程中经常遇到在一个流程中需要调用多个子调用的情况，这些子调用相互之间没有依赖，如果串行调用，则耗时会很长（比如在用户注册后，我们需要发送邮件和短信），此时可以使用`future`模式。

`future`模式的基本工作原理：

（1）使用`chan`作为函数参数

（2）启动`goroutine`调用函数

（3）通过`chan`传入参数

（4）做其他可以并行处理的事情

（5）通过`chan`异步获取结果

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    // 一个查询结构体
    // 这里的 sql 和 result 是一个简单的抽象，具体的应用可能是更复杂的数据类型
    type query struct { 
    	// 参数 Channel
    	sql chan string
    
    	// 结果 Channel
    	result chan string
    }
    
    // 执行 Query
    func execQuery(q query)  { 
    	// 启动协程
    	go func() { 
    		// 获取输入
    		sql := <- q.sql
    
    		// 方法数据库
    
    		// 输入结果通道
    		q.result <- "result from " + sql
    	}()
    }
    
    func main() { 
    	// 初始化 Query
    	q := query{ make(chan string, 1), make(chan string, 1)}
    
    	// 执行 Query
    	go execQuery(q)
    
    	// 发送参数
    	q.sql <- "slect * from table"
    
    	// 做其他事情，通过 sleep 描述
    	time.Sleep(1 * time.Second)
    
    	// 获取结果
    	fmt.Println(<-q.result)
    }

`future`模式最大的好处是将函数的同步调用转化为异步调用。

# 3. context 标准库 #

多个`goroutine`之间的协作工作涉及通信、同步、通知和退出四个方面。

通信：`chan`通道当然是`goroutine`之间通信的基础，这里的通信主要是指程序的数据通道。

同步：不带缓冲的`chan`提供了一个天然的同步等待机制；当然`sync.WaitGroup`也为多个`goroutine`协同工作提供了一种同步等待机制。

通知：通知通常不是业务数据，而是管理、控制流数据。通过在输入端绑定两个`chan`，一个用于业务流数据，另一个用于异常通知数据，然后通过`select`收敛进行处理。

退出：`goroutine` 之间没有负责关系，如果通知`goroutine`退出，可以通过增加一个单独的通道，借助通道和`select`的广播机制实现退出。

## 3.1 context 的设计目的 ##

context 库的设计目的及时跟踪`goroutine`调用树，并在这些`goroutine`调用树中传递通知和元数据。两个目的：

1.  退出通知机制 ---- 通知可以传递给整个`goroutine`调用树上的每一个`goroutine`.
2.  传递数据 ---- 数据可以传递给真个`goroutine`调用树上的每一个`goroutine`.

## 3.2 基本数据结构 ##

context 包的整体工作机制：第一个创建 `Context`的`goroutine`被称为`root`节点。`root` 节点负责创建一个实现 `Context` 接口的具体对象，并将该对象作为参数传递到其新拉起的`goroutine`，下游的`goroutine`可以继续封装该对象，再传递到更下游的`goroutine`。`Context`对象再传递的过程中最终形成一个树状的数据结构，这样通过位于`root`节点(树的根节点)的`Context`对象就能遍历整个`Context`对象树，通知和消息就可以通过`root`节点传递出去，实现上游`goroutine`对下游`goroutine`的消息传递。

### 3.2.1 Context 接口 ###

Context 是一个基本接口，所有的 Context 对象都要实现该接口，context 的使用者再调用接口中都使用`Context`作为参数类型。

type Context interface { 
    	
    	// 如果 Context 实现了超时控制，则该方法返回ok true, deanline 为超时时间
    	// 否则 ok 为 false
    	Deadline() (deadline time.Time, ok bool)
    	
    	// 后端被调用的 goroutine 应该监听该方法返回的 chan，以便及时释放资源
    	Done() <- chan struct{ }
    	
    	// Done 返回的 chan 收到通知的时候，才可以访问 Err() 获知因为什么原因被取消
    	Err() error
    	
    	// 可以访问上游的 goroutine 传递给下游 goroutine 的值
    	Value(key interface{ }) interface{ }
    	
    }

### 3.2.2 canceler 接口 ###

`canceler` 接口是一个扩展接口，规定了取消通知的 `Context` 具体类型需要实现的接口。`context`包中的具体类型`* cancelCtx` 和`* timerCtx`都实现了该接口。

// 一个 context 对象如果实现了 canceler 接口，则可以被取消
    
    type canceler interface { 
    	// 创建一个 cancel 接口实例的 goroutine 调用 cancel 方法通知后续创建的 goroutine 退出
    	cancel(removeFromParent bool, err error)
    	
    	// Done 方法返回的 chan 需要后端 goroutine来监听，以便及时退出
    	Done() <-chan struct{ }
    }

### 3.2.3 empty Context 结构 ###

`emptyCtx`实现了 `Context`接口，但不具备任何功能，因为其所有的方法都是空实现。其存在的目的是作为 `Context`对象树的根节点。

// emptyCtx 实现了 Context 接口
    
    type emptyCtx int
    
    func (*emptyCtx) Deadline() (deadline time.Time, ok bool) { 
    	return
    }
    
    func (*emptyCtx) Done() <-chan struct{ } { 
    	return nil
    }
    
    func (*emptyCtx) Err() error { 
    	return nil
    }
    
    func (*emptyCtx) Value(key interface{ }) interface{ } { 
    	return nil
    }
    
    func (e *emptyCtx) String() string { 
    	switch e { 
    	case background:
    		return "context.Background"
    	case todo:
    		return "context.TODO"
    	}
    	return "unknown empty Context"
    }

package 定义了两个全局遍历和两个封装函数，返回两个 `empryCtx`实例对象，实际使用时通过调用这两个封装函数来构造`Context`的`root`节点。

var (
    	background = new(emptyCtx)
    	todo       = new(emptyCtx)
    )
    
    func Background() Context { 
    	return background
    }
    
    func TODO() Context { 
    	return todo
    }

### 3.2.4 cancelCtx ###

`cancelCtx`是一个实现了`Context`接口的具体类型，同时实现了`conceler`接口。`conceler`具有退出通知方法。退出通知机制不但能通知自己，也能逐层通知其`children`节点。

// cancelCtx 可以被取消，cancelCtx 取消时会同时需要所有实现 canceler 接口的孩子节点
    type cancelCtx struct { 
    	Context
    
    	mu       sync.Mutex            // protects following fields
    	done     chan struct{ }         // created lazily, closed by first cancel call
    	children map[canceler]struct{ } // set to nil by the first cancel call
    	err      error                 // set to non-nil by the first cancel call
    }
    
    func (c *cancelCtx) Done() <-chan struct{ } { 
    	c.mu.Lock()
    	if c.done == nil { 
    		c.done = make(chan struct{ })
    	}
    	d := c.done
    	c.mu.Unlock()
    	return d
    }
    
    func (c *cancelCtx) Err() error { 
    	c.mu.Lock()
    	err := c.err
    	c.mu.Unlock()
    	return err
    }
    
    func (c *cancelCtx) String() string { 
    	return contextName(c.Context) + ".WithCancel"
    }
    
    // cancel closes c.done, cancels each of c's children, and, if
    // removeFromParent is true, removes c from its parent's children.
    func (c *cancelCtx) cancel(removeFromParent bool, err error) { 
    	if err == nil { 
    		panic("context: internal error: missing cancel error")
    	}
    	c.mu.Lock()
    	if c.err != nil { 
    		c.mu.Unlock()
    		return // already canceled
    	}
    	c.err = err
    	if c.done == nil { 
    		c.done = closedchan
    	} else { 
    		// 显式通知自己
    		close(c.done)
    	}
    	// 循环调用 children 的 cancel 函数，由于 parent 已经去洗洗奥，所以此时 child 调用
    	// cancel 传入的是false
    	for child := range c.children { 
    		// NOTE: acquiring the child's lock while holding parent's lock.
    		child.cancel(false, err)
    	}
    	c.children = nil
    	c.mu.Unlock()
    
    	if removeFromParent { 
    		removeChild(c.Context, c)
    	}
    }

### 3.2.5 timerCtx ###

`timerCtx`是一个实现了`Context`接口的具体类型，内部封装了`cancelCtx`类型实例，同时有一个 `deadline`变量，用来实现定时退出通知。

type timerCtx struct { 
    	cancelCtx
    	timer *time.Timer // Under cancelCtx.mu.
    
    	deadline time.Time
    }
    
    func (c *timerCtx) Deadline() (deadline time.Time, ok bool) { 
    	return c.deadline, true
    }
    
    func (c *timerCtx) String() string { 
    	return contextName(c.cancelCtx.Context) + ".WithDeadline(" +
    		c.deadline.String() + " [" +
    		time.Until(c.deadline).String() + "])"
    }
    
    func (c *timerCtx) cancel(removeFromParent bool, err error) { 
    	c.cancelCtx.cancel(false, err)
    	if removeFromParent { 
    		// Remove this timerCtx from its parent cancelCtx's children.
    		removeChild(c.cancelCtx.Context, c)
    	}
    	c.mu.Lock()
    	if c.timer != nil { 
    		c.timer.Stop()
    		c.timer = nil
    	}
    	c.mu.Unlock()
    }

### 3.2.6 valueCtx ###

`valueCtx`是一个实现了`Context`接口的具体类型，内部封装了`Context`接口类型，同时封装了一个`k/v`的存储变量。`valueCtx`可用来传递通知信息。

type valueCtx struct { 
    	Context
    	key, val interface{ }
    }
    
    func (c *valueCtx) String() string { 
    	return contextName(c.Context) + ".WithValue(type " +
    		reflectlite.TypeOf(c.key).String() +
    		", val " + stringify(c.val) + ")"
    }
    
    func (c *valueCtx) Value(key interface{ }) interface{ } { 
    	if c.key == key { 
    		return c.val
    	}
    	return c.Context.Value(key)
    }

## 3.3 API 函数 ##

下面这两个函数是构造 `Context`取消树的根节点对象，根节点对象用作后续`With`包装函数的实参。

func Backgroud() Context
    func TODO() Context

`With`包装函数用来创建不同功能的`Context`具体对象。

1.  创建一个带有退出通知的`Context`具体对象，内部创建一个`cancelCtx`的类型实例。

func WithCancel(parent Context) (ctx Context, cancel CancelFunc) { 
    	c := newCancelCtx(parent)
    	propagateCancel(parent, &c)
    	return &c, func() {  c.cancel(true, Canceled) }
    }

1.  创建一个带有超时通知的`Context`具体对象，内部创建一个`timerCtx`的类型实例。

func WithDeadline(parent Context, d time.Time) (Context, CancelFunc) { 
    	if cur, ok := parent.Deadline(); ok && cur.Before(d) { 
    		// The current deadline is already sooner than the new one.
    		return WithCancel(parent)
    	}
    	c := &timerCtx{ 
    		cancelCtx: newCancelCtx(parent),
    		deadline:  d,
    	}
    	propagateCancel(parent, c)
    	dur := time.Until(d)
    	if dur <= 0 { 
    		c.cancel(true, DeadlineExceeded) // deadline has already passed
    		return c, func() {  c.cancel(false, Canceled) }
    	}
    	c.mu.Lock()
    	defer c.mu.Unlock()
    	if c.err == nil { 
    		c.timer = time.AfterFunc(dur, func() { 
    			c.cancel(true, DeadlineExceeded)
    		})
    	}
    	return c, func() {  c.cancel(true, Canceled) }
    }

1.  创建一个带有超时通知的`Context`具体对象，内部创建一个`timerCtx`的类型实例。

func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) { 
    	return WithDeadline(parent, time.Now().Add(timeout))
    }

1.  创建一个能够传递数据的`Context`具体对象，内部创建一个`valueCtx`的类型实例。

func WithValue(parent Context, key, val interface{ }) Context { 
    	if key == nil { 
    		panic("nil key")
    	}
    	if !reflectlite.TypeOf(key).Comparable() { 
    		panic("key is not comparable")
    	}
    	return &valueCtx{ parent, key, val}
    }

`parent`参数，是实现`Context`通知树的必备条件。在 `goroutine`的调用连中，`Context`的实例被逐层地包装并传递，每层又可以对传进来地`Context`实例再封装自己所需地功能，整个调用树需要一个数据结构来维护，这个维护逻辑再这些包装函数内部实现。

## 3.4 辅助函数 ##

`Context`具体对象地链条关系是在 `With`函数的内部维护的，`With`函数内部使用的通用函数有如下：

func propagateCancel(parent Context, child canceler) { 
    	if parent.Done() == nil { 
    		return // parent is never canceled
    	}
    	if p, ok := parentCancelCtx(parent); ok { 
    		p.mu.Lock()
    		if p.err != nil { 
    			// parent has already been canceled
    			child.cancel(false, p.err)
    		} else { 
    			if p.children == nil { 
    				p.children = make(map[canceler]struct{ })
    			}
    			p.children[child] = struct{ }{ }
    		}
    		p.mu.Unlock()
    	} else { 
    		go func() { 
    			select { 
    			case <-parent.Done():
    				child.cancel(false, parent.Err())
    			case <-child.Done():
    			}
    		}()
    	}
    }

1.  判断 parent 的方法 Done() 返回值是否是 nil，如果是，则说明 parent 不是一个可取消的 Context 对象，也就无所谓取消构造树，说明 child 就是取消构造树的根。
2.  判断 parent 的方法 Done() 返回值不是 nil，则向上回溯自己的祖先是否是`cancelCtx`类型实例，如果是，则将 child 的子节点注册维护到那棵关系树里面。
3.  如果向上回溯自己的祖先都不是`cancelCtx`类型实例，则说明整个链条的取消树是不连续的。此时只需要监听`parent`和自己的取消信号即可。

func parentCancelCtx(parent Context) (*cancelCtx, bool) { 
    	for { 
    		switch c := parent.(type) { 
    		case *cancelCtx:
    			return c, true
    		case *timerCtx:
    			return &c.cancelCtx, true
    		case *valueCtx:
    			parent = c.Context
    		default:
    			return nil, false
    		}
    	}
    }

*  判断 parent 中是否封装了`*cancelCtx`的字段，或者接口里面存放的底层类型是否是`* cancelCtx`类型。

func removeChild(parent Context, child canceler) { 
    	p, ok := parentCancelCtx(parent)
    	if !ok { 
    		return
    	}
    	p.mu.Lock()
    	if p.children != nil { 
    		delete(p.children, child)
    	}
    	p.mu.Unlock()
    }

*  如果 parent封装了`*cancelCtx`类型字段，或者接口里面存放的底层类型是`*cancelCtx类型`，则将其构造树上的`child`节点删除。

## 3.5 context 的用法 ##

package main
    
    import (
    	"context"
    	"fmt"
    	"time"
    )
    
    type otherContext struct { 
    	context.Context
    }
    
    func main() { 
    	// 使用 context.Background() 构建一个 WithCancel 类型的上下文
    	ctxa, cancel := context.WithCancel(context.Background())
    
    	// work 模拟运行并检测前端的退出通知
    	go work(ctxa, "work1")
    
    	// 使用 WithDeadline 包装前面的上下文对象 ctxa
    	tm := time.Now().Add(3 * time.Second)
    	ctxb, _ := context.WithDeadline(ctxa, tm)
    
    	go work(ctxb, "work2")
    
    	// 使用 WithValue 包装前面的上下文对象 ctxb
    	oc := otherContext{ ctxb}
    	ctxc := context.WithValue(oc, "key", "andes,pass from main")
    
    	go workWithValue(ctxc, "work3")
    
    	// sleep 10 秒，让 work2、work3 超时退出
    	time.Sleep(10 * time.Second)
    
    	// 显式调用 work1 的 cancel 方法通知其退出
    	cancel()
    
    	// 等待 work1 打印退出信息
    	time.Sleep(5 * time.Second)
    	fmt.Println("main stop")
    }
    
    func work(ctx context.Context, name string)  { 
    	for{ 
    		select { 
    			case <- ctx.Done():
    				fmt.Printf("%s get msg to cancel\n", name)
    				return
    			default:
    				fmt.Printf("%s is running\n", name)
    				time.Sleep(1 * time.Second)
    		}
    	}
    }
    
    // 等待前端的退出通知，并试图获取 Context 传递的数据
    func workWithValue(ctx context.Context, name string)  { 
    	for{ 
    		select { 
    			case <- ctx.Done():
    				fmt.Printf("%s get msg to cancel\n", name)
    				return
    			default:
    				value := ctx.Value("key").(string)
    				fmt.Printf("%s is running value = %s\n", name, value)
    				time.Sleep(1 * time.Second)
    		}
    	}
    }

// work3 在运行中能够获取前端传递过来的参数 key
    work3 is running value = andes,pass from main
    work1 is running
    work3 is running value = andes,pass from main
    work1 is running
    work2 is running
    work3 is running value = andes,pass from main
    work2 is running
    work1 is running
    // work3 超时退出
    work3 get msg to cancel
    work1 is running
    // work2 超时退出
    work2 get msg to cancel
    work1 is running
    work1 is running
    work1 is running
    work1 is running
    work1 is running
    work1 is running
    // work1 被显式的通知退出
    work1 get msg to cancel
    main stop

**程序分析**

在使用 Context 的过程中，程序在底层实际上维护了两条关系链，两条引用关系链如下。

（1）children key 构成从根到叶子`Context`实例的引用关系，这个关系在调用`With`函数时进行维护，程序有一层这样呢的树状结构：

ctxa.children ---> ctxb
    ctxb.children ---> ctxc

取消通知沿着这条链从根节点向下层节点逐层广播，当然也可以在任意一个子树上调用取消通知，一样会扩散到整棵树。`ctxa`收到退出通知，会通知其绑定的`work1`，通知会广播给`ctxb`和`ctxc`绑定的`work2`和`work3`。同理，`ctxc`收到退出通知，会通知到其绑定的`work2`，同时会广播给`ctxc`绑定的`work3`。

（2）在构造 `Context`的对象中不断包裹`Context`实例形成一个引用关系链，这个关系链的方向是相反的，是自底向上的。

ctxc.Context --> oc
    ctxc.Context.Context --> ctxb
    ctxc.Context.Context.cancelCtx --> ctxa
    ctxc.Context.Context.cancelCtx.Context --> new(EmptyCtx)

这个关系链主要用来切断当前`Context`实例和上层`Context`实例之间的关系，比如`ctxb`调用了退出通知或定时器到期了，`ctxb`后续就没有必要再通知广播树上继续存在，会将自己从广播树清理掉。

`Context`包的一般使用流程如下：

1.  创建一个 Context 根对象
2.  通过 With 函数包装上一步创建的 Context 对象，使其具有特定的功能
3.  将上一步创建的对象作为实参传给后续启动的并发函数，每个并发函数内部可以继续使用包装函数对传进来的 Context 对象进行包装，添加自己所需的功能。
4.  顶端的`goroutine`在超时后调用`cancel`退出通知函数，通知后端的所有`goroutine`释放资源
5.  后端的`goroutine`通过`select`监听`Context.Done()`返回的`chan`，及时响应前端的`goroutine`的退出通知，一般停止本次处理，释放所占用的资源。

# 4. 并发模型 #

## 4.1 CSP 简介 ##

CSP基本思想：将并发系统抽象为`Channel`和`Process`两部分，`Channel`用来传递消息，`Process`用于执行，`Channel`和`Process`之间相互独立，没有从属关系，消息的发送和接收有严格的时序限制。Go 语言主要借鉴了`Channel`和`Process`的概念，在 Go 中`Channel`就是通道，`Process`就是`goroutine`。

## 4.2 调度模型 ##

应用程序的并发模型是多种的，有三种：

*  多进程模型

进程都能多核 CPU 并发调度，优点是每个进程都有自己独立的内存空间，隔离性好、健壮性高；缺点是进程比较重，进程的切换消耗较大，进程间的通信需要多次在内核区和用户区之间复制数据。

*  多线程模型

这里的多线程是指启动多个内核线程进行处理，线程的优点是通过共享内存进行通信更快捷，切换代价小；缺点是多个线程共享内存空间，极易导致数据访问混乱，某个线程误操作内存挂掉可能危及整个进程组，健壮性不高。

*  用户级多线程模型

用户级多进程又分两种情况，一种是`M:1`的方式，M个用户线程对应一个内核线程，这种情况很容易因为一个系统阻塞，导致其他用户线程都会被阻塞，不能利用机器多核的优势。还有一种模式就是`M:N`的方式，M个用户线程对应`N`个内核线程，这种模式一般需要语言运行时或库的支持，效率最高。

协程是一种用户态的轻量级线程，协程的调度完全由用户态程序控制，协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈，内个内核线程可以对应多个用户协程，当一个协程执行体阻塞了，调度器会调度另一个协程执行，最大效率地利用操作系统分给系统线程地时间片。

好处：

1.  控制了系统线程数，保证每个线程地运行时间片充足
2.  调度层能进行用户态地切换，不会导致单个协程阻塞整个程序的情况，尽量减少上下文切换，提升运行效率。

## 4.3 并发和调度 ##

Go 语言在语言层面引入`goroutine`，有以下好处：

1.  `goroutine`可以在用户空间调度，避免了内核态和用户态的切换导致的成本。
2.  `goroutine`是语言原生支持的，提供了非常简洁的语法，屏蔽了大部分复杂底层实现。
3.  `goroutine`更小的栈空间允许用户创建成千上万的实例。

Go 的调度模型重抽象出三个实体：M、P、G

### 4.3.1 G（Goroutine） ###

G 是 Go 运行时对`goroutine`的抽象描述，G 中存放并发执行的代码入口地址、上下文、运行环境（管理的 P 和 M）、运行栈等执行相关的元信息。

G 的新建、休眠、恢复、停止都受到 Go 运行时管理。Go 运行时的监控线程会监控 G 的调度，G 不会长久地阻塞系统线程，运行时的调度器会自动切换到其他 G 上继续运行。G 的新建或恢复时会添加到运行队列，等待 M 取出并运行。

### 4.3.2 M（Machine） ###

M 代表 OS 内核线程，是操作系系统层面调度和执行的实体。M 仅负责执行，M 不停地被唤醒或创建，然后执行。M 启动时进入的是运行时的管理代码，由这段代码获取 G 和 P 资源，然后执行调度。另外，Go 语言运行时会单独创建一个监控线程，负责对程序的内存、调度等信息进行监控和控制。

### 4.3.3 P（Processor） ###

P 代表 M 运行 G 所需要的资源，是对资源的一种抽象和管理，P 不是一段代码实体，而是一个管理的数据结构，P 主要是降低 M 管理调度 G 的复杂性，增加一个间接的控制层数据结构。把 P 看作资源，而不是处理器，P 控制 Go 代码的并行读，它不是运行实体。P 持有 G 的队列，P 可以隔离调度，解除 P 和 M 的绑定就解除了 M 对 一串 G 的调用。 P 在运行模型中只是一个数据模型，而不是程序控制模型。

M 和 P 一起构成一个运行时环境，每个 P 有一个本地的可调度 G 队列，队列里面的 G 会被 M 依次 调度执行，如果本地队列为空了，则会去全局队列偷取一部分 G，如果全局队列也是空的，则去其他的 P 中偷取一部分 G，这就是 Work Stealing 算法的基本原理。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 8]  
      G 并不是执行体，而是用于存放并发执行体的元信息，包括并发执行的入口函数、堆栈上下文等信息。G 由于保存的是元信息，为了减少对象的分配和回收，G 对象是可以复用的，只需将相关元信息初始化为新值即可。M 仅负责执行，M 启动时进入运行时的管理代码，这段管理代码必须拿到可用的 P 后，才能执行调度。P 的数目默认时 CPu 核心数量，可以通过 `runtime.GOMAXPROCS`函数设置或查询，M 和 P 的数目差不多，但运行时会根据当前的状态创建 M，M 有一个最大值上限，目前时 10000；G 与 P 是一种 M : N 的惯性系，M 可以成千上万，远远大于 N。

**m0和g0**

Go 中还有特殊的 M 和 G，它们是 m0 和 g0。m0 时候启动程序后的主线程，这个 m 对象的信息会存放在全局变量 m0 中，m0 负责执行初始化操作和启动第一个 g，之后 m0 就和其他的 M 一样了。

每个 M 都会有一个自己的管理堆栈 g0，g0 不指向任何可指向的函数，g0 仅在 M 执行管理的调度逻辑时使用。在调度或系统调用时会切换到 g0 的栈空间，全局变量的 g0 是 m0 的 g0。

**Go 启动初始化过程**

1.  分配和检查栈空间
2.  初始化参数和环境变量
3.  当前运行线程标记为 m0，m0 是程序启动的主线程
4.  调用运行时初始化函数`runtime.schedinit`进行初始化（主要是初始化内存空间分配器、GC、生成空闲 P 列表）
5.  在 m0 上调度第一个 G，这个 G 运行`runtime.main`函数。（`runtime.main()`函数会拉起运行时的监控线程，然后调用`main`包的`init()`初始化函数，最后执行`main`函数）

**什么时候创建 M、P、G**

在程序启动过程中会初始化空闲 P 列表，P 是在这个时候被创建的，同时第一个 G 也是在初始化过程中被创建的，后续在有 go 并发调用的地方都有可能创建 G，由于 G 只是一个数据结构，并不是执行实体，所以 G 是可以被复用的。在需要 G 结构时，首先要去 P 的空闲 G 列表里面寻找已经运行结束的 `goroutine`，其 G 会被缓存起来。

每个并发调用都会初始化一个新的 G 任务，然后唤醒 M 执行任务。这个唤醒不是特定唤醒某个线程去工作，而是先尝试获取当前线程 M，如果无法获取，则从全局调度的空闲 M 列表中获取可用的 M，如果没有可用的，则新建 M，然后绑定 P 和 G 进行运行。所以 M 和 P 不是一一对应的，M 是按需分配的，但是运行时会设置一个上限值（默认时 10000），超出最大值将导致程序崩溃。

**抢占调度的策略**

1.  在进入系统调用前后，各封装一层代码检测 G 的状态，当检测到当前 G 已经被监控线程抢占调度，则 M 停止执行当前 G，进行调度切换。
2.  监控线程经过一段时间检测感知到 P 运行超过一定时间，取消 P 和 M 的关联，这也是一种更高层次的调度。
3.  监控线程经过一段时间检测感知到 G 一直运行，超过了一定的时间，设置 G 标记，G 执行栈扩展逻辑检测到抢占标记，根据相关条件决定是否抢占调度。

[20200517094109609.png]: /images/20210813/b97724527d3a46aa85af229e5d301191.png
[20200517094201763.png]: /images/20210813/e1fea6f75fb84a89bbc79cb9ae1dcf46.png
[2020051709561626.png]: /images/20210813/d090bbef70fd422c92adc57579a80d50.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70]: /images/20210813/cd859adbc7e64d1fa7c8034a32e3533a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 1]: /images/20210813/bd3b4016097e4800a087f5b1e27f1268.png
[20200517105437817.png]: /images/20210813/ac557739c715482c899465c390700169.png
[20200517105448890.png]: /images/20210813/d437c1717b77424e99a7e4c69204db0f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 2]: /images/20210813/b268f894dbe444cd8728b4327d7ed198.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 3]: /images/20210813/f6b1a202003f4b0db235956e17bfc019.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20200517110226990.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 5]: /images/20210813/fb44b07c37c04d8387e1bd23e428e732.png
[20200517115511911.png]: /images/20210813/b3b1edf5a6804d17be245c83585a3c8e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 6]: /images/20210813/eedd8b9c4ab942fbbdae1f004845a1d0.png
[20200517194911265.png]: /images/20210813/d62a1bd00326405082dab5c19d48ccdd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 7]: /images/20210813/e9188c407050441db926e9a3afc7c1ef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdfemhhb18_size_16_color_FFFFFF_t_70 8]: /images/20210813/c145e54eb073466aab439e684147cb02.png