协程 (goroutine) 与通道 (channel)
Go 语言为构建并发程序的基本代码块是协程 (goroutine) 与通道 (channel)。他们需要语言，编译器，和 runtime 的支持。Go 语言提供的垃圾回收器对并发编程至关重要。
不要通过共享内存来通信，而通过通信来共享内存。
通信强制协作。
并发、并行和协程
一个应用程序是运行在机器上的一个进程；进程是一个运行在自己内存地址空间里的独立执行体。一个进程由一个或多个操作系统线程组成，这些线程其实是共享同一个内存地址空间的一起工作的执行体。
并行是一种通过使用多处理器以提高速度的能力。所以并发程序可以是并行的，也可以不是。
并发程序在一个单个处理器上 使用多个线程 执行任务，只是宏观的并行，微观是串行，方式可能为交替或是其他规则的调度，只有多核处理器或多处理器，才是真正的并行
多线程并发，难以做到准确，最主要的问题是内存中的数据共享，它们会被多线程以无法预知的方式进行操作，导致一些无法重现或者随机的结果（称作竞态）
不要使用全局变量或者共享内存，它们会给你的代码在并发运算的时候带来危险。
解决之道在于同步不同的线程，对数据加锁，这样同时就只有一个线程可以变更数据。(标准库 sync)
Go 更倾向于其他的方式，在诸多比较合适的范式中，有个被称作 Communicating Sequential Processes（顺序通信处理）（CSP, C. Hoare 发明的）还有一个叫做 message passing-model（消息传递）（已经运用在了其他语言中，比如 Erlang）。
在 Go 中，应用程序并发处理的部分被称作 goroutines（协程），它可以进行更有效的并发运算。在协程和操作系统线程之间并无一对一的关系：协程是根据一个或多个线程的可用性，映射（多路复用，执行于）在他们之上的；协程调度器在 Go 运行时很好的完成了这个工作。
当系统调用（比如等待 I/O）阻塞协程时，其他协程会继续在其他线程上工作。协程的设计隐藏了许多线程创建和管理方面的复杂工作。(大概是这个堵塞时，其他运行于此的协程就会改映射，通过别的线程继续运行？)
协程是轻量的，比线程更轻。它们痕迹非常不明显（使用少量的内存和资源）：使用 4K 的栈内存就可以在堆中创建它们（4K栈可以管理多少个协程？）。因为创建非常廉价，必要的时候可以轻松创建并运行大量的协程（在同一个地址空间中 100,000 个连续的协程）。并且它们对栈进行了分割，从而动态的增加（或缩减）内存的使用；栈的管理是自动的，但不是由垃圾回收器管理的，而是在协程退出后自动释放。
协程可以运行在多个操作系统线程之间，也可以运行在线程之内。由于操作系统线程上的协程时间片，你可以使用少量的操作系统线程就能拥有任意多个提供服务的协程。Go 运行时可以聪明的意识到哪些协程被阻塞了，暂时搁置它们并处理其他协程。
协程是通过使用关键字 go 调用（执行）一个函数或者方法来实现的（也可以是匿名或者 lambda 函数）。这样会在当前的计算过程中开始一个同时进行的函数，在相同的地址空间中（进行）并且分配了独立的栈，比如：go sum(bigArray)，在后台计算总和。
协程的栈会根据需要进行伸缩，不出现栈溢出；开发者不需要关心栈的大小。当协程结束的时候，它会静默退出：用来启动这个协程的函数不会得到任何的返回值。
任何 Go 程序都必须有的 main() 函数也可以看做是一个协程，尽管它并没有通过 go 来启动。协程可以在程序初始化的过程中运行（在 init() 函数中）。
在一个协程中，比如它需要进行非常密集的运算，你可以在运算循环中周期的使用 runtime.Gosched()：这会让出处理器，允许运行其他协程；它并不会使当前协程挂起，所以它会自动恢复执行。使用 Gosched() 可以使计算均匀分布，使通信不至于迟迟得不到响应。（意思是，你吃饭很多，隔一段时间让一个人吃一口，然后自己继续吃，不至于，在后面排队的人，等了很久一口饭都没吃，而你却吃到饱。）
并发和并行的差异
在 gc 编译器下（6g 或者 8g）你必须设置 GOMAXPROCS 为一个大于默认值 1 的数值来允许运行时支持使用多于 1 个的操作系统线程，所有的协程都会共享同一个线程除非将 GOMAXPROCS 设置为一个大于 1 的数。当 GOMAXPROCS 大于 1 时，会有一个线程池管理许多的线程。
。如果你设置环境变量 GOMAXPROCS>=n，或者执行 runtime.GOMAXPROCS(n)，接下来协程会被分割（分散）到 n 个处理器上。
有这样一个经验法则，对于 n 个核心的情况设置 GOMAXPROCS 为 n-1 以获得最佳性能，也同样需要遵守这条规则：协程的数量 > 1 + GOMAXPROCS > 1。
所以如果在某一时间只有一个协程在执行，不要设置 GOMAXPROCS！
经验总结：适当增加GOMAXPROCS是有好处的，但是不能太大，否则事与愿违，要留一个核心，其他全分配
package main
import (
"fmt"
"runtime"
"time"
)
func main() {
// setting GOMAXPROCS to 2 gives +- 22% performance increase,
// but increasing the number doesn't increase the performance
// without GOMAXPROCS: +- 86000
// setting GOMAXPROCS to 2: +- 105000
// setting GOMAXPROCS to 3: +- 94000
runtime.GOMAXPROCS(2)
ch1 := make(chan int)
ch2 := make(chan int)
go pump1(ch1)
go pump2(ch2)
go suck(ch1, ch2)
time.Sleep(1e9)
}
func pump1(ch chan int) {
for i := 0; ; i++ {
ch <- i * 2
}
}
func pump2(ch chan int) {
for i := 0; ; i++ {
ch <- i + 5
}
}
func suck(ch1, ch2 chan int) {
for i := 0; ; i++ {
select {
case v := <-ch1:
fmt.Printf("%d - Received on channel 1: %d\n", i, v)
case v := <-ch2:
fmt.Printf("%d - Received on channel 2: %d\n", i, v)
}
}
}
总结：GOMAXPROCS 等同于（并发的）线程数量，在一台核心数多于 1 个的机器上，会尽可能有等同于核心数的线程在并行运行。（真正的瓶颈在于核心数，超过核心数的并发实际上是宏观并行，微观串行，而线程数=核心数时，会尽力保证都是 微观并行（毕竟除了本程序还有其他程序在用CPU），后者才是真正的性能提升）
如何用命令行指定使用的核心数量
使用 flags 包，如下：
var numCores = flag.Int("n", 2, "number of CPU cores to use")
在 main() 中：
flag.Parse()
runtime.GOMAXPROCS(*numCores)
协程可以通过调用 runtime.Goexit() 来停止
package main
import (
"fmt"
"time"
)
func main() {
fmt.Println("In main()")
go longWait() // 使用协程
go shortWait()
fmt.Println("About to sleep in main()")
// sleep works with a Duration in nanoseconds (ns) !
time.Sleep(10 * 1e9)
fmt.Println("At the end of main()")
}
func longWait() {
fmt.Println("Beginning longWait()")
time.Sleep(5 * 1e9) // sleep for 5 seconds
fmt.Println("End of longWait()")
}
func shortWait() {
fmt.Println("Beginning shortWait()")
time.Sleep(2 * 1e9) // sleep for 2 seconds
fmt.Println("End of shortWait()")
}
main()，longWait() 和 shortWait() 三个函数作为独立的处理单元按顺序启动，然后开始并行运行。
Sleep() 可以按照指定的时间来暂停函数或协程的执行，这里使用了纳秒（ns，符号 1e9 表示 1 乘 10 的 9 次方，e=指数）
当 main() 函数返回的时候，程序退出：它不会等待任何其他非 main 协程的结束。这就是为什么在服务器程序中，每一个请求都会启动一个协程来处理，server() 函数必须保持运行状态。通常使用一个无限循环来达到这样的目的。(程序中，main为10秒，各并行程序最长的为5秒，所以能正常运行)
为了对比使用一个线程，连续调用的情况，移除 go 关键字，重新运行程序。
现在输出：
In main()
Beginning longWait()
End of longWait()
Beginning shortWait()
End of shortWait()
About to sleep in main()
At the end of main() // after 17 s
协程的经典作用
协程更有用的一个例子应该是在一个非常长的数组中查找一个元素。
将数组分割为若干个不重复的切片，然后给每一个切片启动一个协程进行查找计算。这样许多并行的协程可以用来进行查找任务，整体的查找时间会缩短（除以协程的数量）。
Go 协程 (goroutines) 和协程 (coroutines)
在其他语言中，比如 C#，Lua 或者 Python 都有协程的概念。这个名字表明它和 Go 协程有些相似，不过有两点不同：
Go 协程意味着并行（或者可以以并行的方式部署），协程一般来说不是这样的
Go 协程通过通道来通信；协程通过让出和恢复操作来通信
Go 协程比协程更强大，也很容易从协程的逻辑复用到 Go 协程。
协程间的信道
协程可以使用共享变量来通信，但是很不提倡这样做，因为这种方式给所有的共享内存的多线程都带来了困难。（即共享内存的同步方式，传统方式）
通道（channel）
数据在通道中进行传递：在任何给定时间，一个数据被设计为只有一个协程可以对其访问，所以不会发生数据竞争。 数据的所有权（可以读写数据的能力）也因此被传递。
声明
var identifier chan datatype
未初始化的通道的值是 nil
通过datatype知道通道 是区分数据类型的（允许类型检查），所有的类型都可以用于通道，空接口 interface{} 也可以，甚至可以（有时非常有用）创建通道的通道。
通道的本质
通道实际上是类型化消息的队列：使数据得以传输。它是先进先出(FIFO) 的结构所以可以保证发送给他们的元素的顺序，通道创建的时候都是双向的（有些人知道，通道可以比作 Unix shells 中的双向管道 (two-way pipe) ）。通
道也是引用类型，所以我们使用 make() 函数来给它分配内存。
var ch1 chan string // 创建通道（变量）
ch1 = make(chan string)// 实例化
或
ch1 := make(chan string) // 一步到位
或 通道的通道
chanOfChans := make(chan chan int)
通信操作符 <-
这个操作符直观的标示了数据的传输：信息按照箭头的方向流动
流向通道（发送）
ch <- int1
通道的神奇特性
package main
import "fmt"
func main() {
stopCh := make(chan int)
close(stopCh)
select {
case <-stopCh:
fmt.Println("ok")
}
}
输出 ok
close通道会导致<-stopCh的解堵塞。
从通道流出（接收）
int2 = <- ch
int2 := <- ch //int2 未声明
// <- ch 可以单独调用获取通道的（下一个）值
if <- ch != 1000{
...
}
读写互不干扰
通道的发送和接收都是原子操作：它们总是互不干扰地完成
getData() 使用了无限循环：它随着 sendData() 的发送完成和 ch 变空也结束了
main() 等待了 1 秒让两个协程完成，如果不这样，sendData() 就没有机会输出
package main
import (
"fmt"
"time"
)
func main() { //会等待1秒
ch := make(chan string)
go sendData(ch)
go getData(ch)
time.Sleep(1e9)
}
func sendData(ch chan string) {
ch <- "Washington"
ch <- "Tripoli"
ch <- "London"
ch <- "Beijing"
ch <- "Tokyo"
}
func getData(ch chan string) {
var input string
// time.Sleep(2e9)
for {
input = <-ch //一个个取出来，取到空结束
fmt.Printf("%s ", input)
}
}
注意：不要使用打印状态来表明通道的发送和接收顺序：由于打印状态和通道实际发生读写的时间延迟会导致和真实发生的顺序不同。
如果我们移除一个或所有 go 关键字，程序无法运行，Go 运行时会抛出 panic：死锁（为啥？答案见下面 通道阻塞）
无缓冲通道 - 通道阻塞
默认情况下，通信是同步且无缓冲的：
对于同一个通道，在有接受者接收数据之前，发送不会结束，会一直阻塞着。此时的通道为非空通道，新的输入 是无法在通道非空的情况下传入，直到非空时才开始新的输入。
对于同一个通道，接收操作是阻塞的（协程或函数中的），直到发送者可用：如果通道中没有数据，接收者就阻塞了。
单独调用（<- ch）会解除阻塞，但是只能解除一次，除非对fmt.Println(<-ch1)使用循环调用
package main
import "fmt"
func main() {
ch1 := make(chan int)
go pump(ch1) // pump hangs
fmt.Println(<-ch1) // prints only 0
}
func pump(ch chan int) {
for i := 0; ; i++ {
ch <- i
}
}
输出：0
以下程序会发生死锁
package main
import (
"fmt"
)
func f1(in chan int) {
fmt.Println(<-in)
}
func main() {
out := make(chan int)
out <- 2
go f1(out)
}
在 非并发中 执行 out <- 2 ，由于没有抢占，通道内的值 永远不可能 被使用 ，所以out <- 2一直阻塞着（是否死锁，重点看 阻塞的是不是 主协程，也就是main协程）
补充：
https://blog.csdn.net/a13545564067/article/details/115065658
package main
var a string
func f() {
print(a)
}
func hello() {
a = "hello, world"
go f()
}
func main() {
hello()
}
输出：
不确定
可能调用hello函数的某一时刻打印；可能hello函数执行完成后打印；可能不打印"hello world"
原因：
执行go f()语句创建Goroutine和hello函数是在同一个Goroutine中执行, 根据语句的书写顺序可以确定Goroutine的创建发生在hello函数返回之前, 但是新创建Goroutine对应的f()的执行事件和hello函数返回的事件则是不可排序的，也就是并发的。
故不确定f()执行打印语句和hello函数返回谁先谁后，所以可能在hello函数返回之前或返回之后打印；如果hello函数返回了且main函数也退出了，那么程序就退出了，打印语句不会再执行。
package main
import (
"fmt"
)
func main() {
ch := make(chan int)
fmt.Println(<-ch)
go func(){
ch <- 1
}()
}
输出：
deadlock!
原因：
main函数执行到第9行时，要从通道ch取值，此时main协程会被阻塞，等待向通道存值的操作；而在main协程中程序是顺序执行的，如果第9行一直阻塞着，下面的go协程也不可能执行，main协程也永远等不来向通道存值的操作；运行时会检查主协程是否在等待，如果是则判定为死锁。
package main
import (
"time"
)
func main() {
ch1 := make(chan string)
ch2 := make(chan string)
go func() {
ch2 <- "ch2 value"
ch1 <- "ch1 value"
}()
go func() {
<- ch1 //注意顺序
<- ch2
}()
time.Sleep(time.Second * 2)
print("main done")
}
输出：
main done
原因：
main函数顺序执行到第8行时，此时摆在面前的有3个协程：2个go协程和main协程，它们的执行是并发的，谁都有可能先发生。main协程第19行休眠2s，保证main协程不会太快执行完，因为如果main函数执行结束，程序会退出，不会等待其他非main协程结束；不管这2个go协程哪个先执行，都会导致2个go协程都被阻塞等待；此时main协程休眠完，执行完第20行，程序退出。尽管2个go协程互相阻塞，但和main协程无关，故不会报死锁错误。
package main
var done = make(chan bool, 1)
var msg string
func aGoroutine() {
msg = "hello, world"
<-done
}
func main() {
go aGoroutine()
done <- true
println(msg)
}
输出：
2种可能：
打印为空或打印"hello world"
原因：
通道done是带缓冲的，main协程的done <- true接收操作将不会被后台协程的<-done接收操作阻塞，若第14行打印语句早于aGoroutine协程对msg赋值执行，则打印为空；若msg赋值早于打印语句执行，则打印"hello world"。
补充：
1、互斥使用，即当资源被一个线程使用(占有)时，别的线程不能使用
2、不可抢占，资源请求者不能强制从资源占有者手中夺取资源，资源只能由资源占有者主动释放。
3、请求和保持，即当资源请求者在请求其他的资源的同时保持对原有资源的占有。
4、循环等待，即存在一个等待队列：P1占有P2的资源，P2占有P3的资源，P3占有P1的资源。这样就形成了一个等待环路。
https://blog.csdn.net/qq_31762741/article/details/123274086
有缓冲通道
创建
buf := 100
ch1 := make(chan string, buf) // buf 是通道可以同时容纳的元素（这里是 string）个数，内置的 cap() 函数可以返回缓冲区的容量。
缓冲满载前，发送数据不会堵塞，缓冲清空前，读取数据也不会堵塞，元素会按照发送的顺序被接收。
同步：ch :=make(chan type, value)
value == 0 -> synchronous, unbuffered （阻塞）
value > 0 -> asynchronous, buffered（非阻塞）取决于 value 元素
若使用通道的缓冲，你的程序会在“请求”激增的时候表现更好：更具弹性，专业术语叫：更具有伸缩性(scalable)。在设计算法时首先考虑使用无缓冲通道，只在不确定的情况下使用缓冲。
协程 - 信号量模式（无缓冲）
通过通道获知是否已经完成操作，从而达到同步效果。如通过通过获知 操作1 是否完成， 以便继续执行操作2，因为操作2必须在操作1之后。
以下代码可以在未知 切片长度的情况下，知道什么时候 整合切片的元素都doSomething 这件事执行完毕，那么为啥 元素doSomething 要并发执行，因为 元素doSomething 并发执行 可以节省时间。
type Empty interface {}
var empty Empty
...
data := make([]float64, N)
res := make([]float64, N)
sem := make(chan Empty, N)
...
for i, xi := range data { //带检测关闭的无限循环
go func (i int, xi float64) {
res[i] = doSomething(i, xi)
sem <- empty
} (i, xi)
}
// wait for goroutines to finish
for i := 0; i < N; i++ { <-sem }
注意上述代码中闭合函数的用法：i、xi 都是作为参数传入闭合函数的，这一做法使得每个协程（译者注：在其启动时）获得一份 i 和 xi 的单独拷贝，从而向闭合函数内部屏蔽了外层循环中的 i 和 xi 变量；（闭包保值）
否则，for 循环的下一次迭代会更新所有协程中 i 和 xi 的值。另一方面，切片 res 没有传入闭合函数，因为协程不需要 res 的单独拷贝。
切片 res 也在闭合函数中但并不是参数，它也不需要单独拷贝，它并不会变。
package main
import "fmt"
// integer producer:
func numGen(start, count int, out chan<- int) { // start其实，count步长
for i := 0; i < count; i++ {
out <- start
start = start + count
}
close(out)
}
// integer consumer:
func numEchoRange(in <-chan int, done chan<- bool) {
for num := range in {
fmt.Printf("%d\n", num)
}
done <- true
}
func main() {
numChan := make(chan int)
done := make(chan bool)
go numGen(0, 10, numChan)
go numEchoRange(numChan, done)
<-done
}
/* Output:
0
10
20
30
40
50
60
70
80
90
*/
协程 - 信号量模式（有缓冲）
Go语言中有缓冲的通道（buffered channel）是一种在被接收前能存储一个或者多个值的通道。这种类型的通道并不强制要求 goroutine 之间必须同时完成发送和接收。通道会阻塞发送和接收动作的条件也会不同。只有在通道中没有要接收的值时，接收动作才会阻塞。只有在通道没有可用缓冲区容纳被发送的值时，发送动作才会阻塞。
这导致有缓冲的通道和无缓冲的通道之间的一个很大的不同：无缓冲的通道保证进行发送和接收的 goroutine 会在同一时间进行数据交换；有缓冲的通道没有这种保证。
http://c.biancheng.net/view/100.html
通道分为两部分：长度和空位
带缓冲通道的容量和要同步的资源容量相同
通道的长度（当前存放的元素个数）与当前资源被使用的数量相同
容量减去通道的长度就是未处理的资源个数（标准信号量的整数值）
翻译
带缓冲通道的容量 就是 empty
通道的长度 就是 full
empty - full = 空位(剩余的)
//生产者进程
while(TRUE)
{
生产一个产品；
P(empty);
将一个产品送入缓冲区in；
in = (in+1) mod n;
V(full);
}
//消费者进程
while(TRUE)
{
P(full);
从缓冲区(out)取出一个产品；
out = (out+1) mod n;
V(empty);
消费取出的产品；
}
Go 实现P V 操作
// acquire n resources
func (s semaphore) P(n int) {
e := new(Empty)
for i := 0; i < n; i++ {
s <- e
}
}
// release n resources
func (s semaphore) V(n int) {
for i:= 0; i < n; i++{
<- s
}
}
/* mutexes */
func (s semaphore) Lock() {
s.P(1)
}
func (s semaphore) Unlock(){
s.V(1)
}
/* signal-wait */
func (s semaphore) Wait(n int) {
s.P(n)
}
func (s semaphore) Signal() {
s.V(1)
}
习惯用法：通道工厂模式
编程中常见的另外一种模式如下：不将通道作为参数传递给协程，而用函数来生成一个通道并返回（工厂角色）；函数内有个匿名函数被协程调用。
package main
import (
"fmt"
"time"
)
func main() {
stream := pump()
go suck(stream)
time.Sleep(1e9)//等待1s,main的Sleep不影响suck
}
func pump() chan int {
ch := make(chan int)
go func() {
for i := 0; ; i++ {
ch <- i
}
}()
return ch
}
func suck(ch chan int) {
for {
fmt.Println(<-ch)
}
}
给通道使用 for 循环
for range+通道，这段程序是不会停止的，直到main苏醒过来
package main
import (
"fmt"
"time"
)
func main() {
suck(pump())
time.Sleep(1e9)
}
func pump() chan int {
ch := make(chan int)
go func() {
for i := 0; ; i++ {
ch <- i
}
}()
return ch
}
func suck(ch chan int) {
go func() {
for v := range ch {
fmt.Println(v)
}
}()
}
习惯用法：通道迭代器模式
func (c *container) Iter () <- chan item {
ch := make(chan item)
go func () {
for i:= 0; i < c.Len(); i++{ // or use a for-range loop
ch <- c.items[i]
}
} ()
return ch
}
for x := range container.Iter() { ... }
习惯用法：生产者消费者模式
for {
Consume(Produce())
}
上面两个差不多，都是循环里调 消费者
通道的方向
单向通道
var send_only chan<- int // channel can only receive data
var recv_only <-chan int // channel can only send data
只接收的通道 (<-chan T) 无法关闭，因为关闭通道是发送者用来表示不再给通道发送值了，所以对只接收通道是没有意义的
通道创建的时候都是双向的，但也可以分配给有方向的通道变量，就像以下代码：
var c = make(chan int) // bidirectional
go source(c)
go sink(c)
func source(ch chan<- int){
for { ch <- 1 }
}
func sink(ch <-chan int) {
for { <-ch }
}
组装管道
sendChan := make(chan int)
receiveChan := make(chan string)
go processChannel(sendChan, receiveChan) // 组装收发方
func processChannel(in <-chan int, out chan<- string) {
for inValue := range in {
result := ... /// processing inValue
out <- result
}
}
习惯用法：管道和选择器模式
第一个例子大概看看就行，主要看第二个例子
第一个例子：
// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.package main
package main
import "fmt"
// Send the sequence 2, 3, 4, ... to channel 'ch'.
func generate(ch chan int) {
for i := 2; ; i++ {
ch <- i // Send 'i' to channel 'ch'.
}
}
// Copy the values from channel 'in' to channel 'out',
// removing those divisible by 'prime'.
func filter(in, out chan int, prime int) {
for {
i := <-in // Receive value of new variable 'i' from 'in'.
if i%prime != 0 {
out <- i // Send 'i' to channel 'out'.
}
}
}
// The prime sieve: Daisy-chain filter processes together.
func main() {
ch := make(chan int) // Create a new channel.
go generate(ch) // Start generate() as a goroutine.
for {
prime := <-ch
fmt.Print(prime, " ")
ch1 := make(chan int)
go filter(ch, ch1, prime)
ch = ch1
}
}
协程 filter(in, out chan int, prime int) 拷贝整数到输出通道，丢弃掉可以被 prime 整除的数字。然后每个 prime 又开启了一个新的协程，生成器和选择器并发请求。
第二个例子：
// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package main
import (
"fmt"
)
// Send the sequence 2, 3, 4, ... to returned channel
func generate() chan int {
ch := make(chan int)
go func() {
for i := 2; ; i++ {
ch <- i
}
}()
return ch
}
// Filter out input values divisible by 'prime', send rest to returned channel
func filter(in chan int, prime int) chan int {
out := make(chan int)
go func() {
for {
if i := <-in; i%prime != 0 {
out <- i
}
}
}()
return out
}
func sieve() chan int {
out := make(chan int)
go func() {
ch := generate()
for {
prime := <-ch
ch = filter(ch, prime)
out <- prime
}
}()
return out
}
func main() {
primes := sieve()
for {
fmt.Println(<-primes)
}
}
第二个版本引入了上边的习惯用法：函数 sieve()、generate() 和 filter() 都是工厂:它们都是创建通道并返回，而且使用了协程的 lambda 函数。
main() 函数现在短小清晰：它调用 sieve() 返回了包含素数的通道，然后通过 fmt.Println(<-primes) 打印出来
以上三个工程，都是并发执行，之间的顺序由通道保证。for + fmt.Println(<-primes) 源源不断地解除通道的阻塞，从而激活整个流程不断重复进行
协程同步：关闭通道-测试阻塞的通道
关闭：
只有发可以关
收 怎么知道是否关了
阻塞：
怎么检测（测试）阻塞
关闭通道的后果：这个将通道标记为无法通过发送操作 <- 接受更多的值，给已经关闭的通道发送或者再次关闭都会导致运行时的 panic()
通常的 关 法
ch := make(chan float64)
defer close(ch)
使用逗号 ok 模式用来检测通道是否被关闭
v, ok := <-ch // ok is true if v received value
或
if v, ok := <-ch; ok {
process(v)
}
或
for{
v, ok := <-ch
if !ok {
break
}
process(v)
}
使用逗号 select 用来检测通道是否被关闭，它的行为像是“你准备好了吗”的轮询机制
select {
case v, ok := <-ch:
if ok {
process(v)
} else {
fmt.Println("The channel is closed")
}
default:
fmt.Println("The channel is blocked")
}
综合运用
package main
import "fmt"
func main() {
ch := make(chan string)
go sendData(ch)
getData(ch)
}
func sendData(ch chan string) {
ch <- "Washington"
ch <- "Tripoli"
ch <- "London"
ch <- "Beijing"
ch <- "Tokio"
close(ch)
}
func getData(ch chan string) {
for {
input, open := <-ch
if !open {
break
}
fmt.Printf("%s ", input)
}
}
这个程序在之前出现过，若getData不是协程时，当sendData不再向通道发送值时，getData就会陷入无限等待，会发生死锁，而此处加了close(ch)+open+break的配合可以马上退出循环，从而打破死锁
使用 for-range 语句来读取通道是更好的办法，因为这会自动检测通道是否关闭：
for input := range ch {
process(input)
}
在第 14.2.10 节的通道迭代器中，两个协程经常是一个阻塞另外一个。如果程序工作在多核心的机器上，大部分时间只用到了一个处理器。可以通过使用带缓冲（缓冲空间大于 0）的通道来改善。比如，缓冲大小为 100，迭代器在阻塞之前，至少可以从容器获得 100 个元素。如果消费者协程在独立的内核运行(真正并行)，就有可能让协程不会出现阻塞。
由于容器中元素的数量通常是已知的，需要让通道有足够的容量放置所有的元素。这样，迭代器就不会阻塞（尽管消费者协程仍然可能阻塞）。然而，这实际上加倍了迭代容器所需要的内存使用量（通道太大，很费内存），所以通道的容量需要限制一下最大值。记录运行时间和性能测试可以帮助你找到最小的缓存容量带来最好的性能。
使用 select 切换协程
从不同的并发执行的协程中获取值可以通过关键字 select 来完成，它和 switch 控制语句非常相似（章节 5.3）也被称作通信开关；
它的行为像是“你准备好了吗”的轮询机制；
select 监听进入通道的数据，也可以是用通道发送值的时候。
select {
case u:= <- ch1:
...
case v:= <- ch2:
...
...
default: // no value ready to be received
...
}
自动break，不能使用fallthrough ，使用break或者return，select 就结束了
执行思路：
select 做的就是：选择处理列出的多个通信情况中的一个。
如果都阻塞了，会等待直到其中一个可以处理
如果多个可以处理，随机选择一个
如果没有通道操作可以处理并且写了 default 语句，它就会执行：default 永远是可运行的（这就是准备好了，可以执行）。
在 select 中使用发送操作并且有 default 可以确保发送不被阻塞！如果没有 default，select 就会一直阻塞。
package main
import (
"fmt"
"runtime"
"time"
)
func main() {
// setting GOMAXPROCS to 2 gives +- 22% performance increase,
// but increasing the number doesn't increase the performance
// without GOMAXPROCS: +- 86000
// setting GOMAXPROCS to 2: +- 105000
// setting GOMAXPROCS to 3: +- 94000
runtime.GOMAXPROCS(2)
ch1 := make(chan int)
ch2 := make(chan int)
go pump1(ch1)
go pump2(ch2)
go suck(ch1, ch2)
time.Sleep(1e9)
}
func pump1(ch chan int) {
for i := 0; ; i++ {
ch <- i * 2
}
}
func pump2(ch chan int) {
for i := 0; ; i++ {
ch <- i + 5
}
}
func suck(ch1, ch2 chan int) {
for i := 0; ; i++ {
select {
case v := <-ch1:
fmt.Printf("%d - Received on channel 1: %d\n", i, v)
case v := <-ch2:
fmt.Printf("%d - Received on channel 2: %d\n", i, v)
}
}
}
习惯用法：后台服务模式
select实现
服务通常是是用后台协程中的无限循环实现的，在循环中使用 select 获取并处理通道中的数据：
// Backend goroutine.
func backend() {
for {
select {
case cmd := <-ch1:
// Handle ...
case cmd := <-ch2:
...
case cmd := <-chStop:
// stop server
}
}
}
在程序的其他地方给通道 ch1，ch2 发送数据，比如：通道 stop 用来清理结束服务程序。
传统实现
另一种方式（但是不太灵活）就是（客户端）在 chRequest 上提交请求，后台协程循环这个通道，使用 switch 根据请求的行为来分别处理：
func backend() {
for req := range chRequest {
switch req.Subjext() {
case A1: // Handle case ...
case A2: // Handle case ...
default:
// Handle illegal request ..
// ...
}
}
}
通道、超时和计时器（Ticker）
Ticker结构体和Ticker工厂函数
time包有一个 结构体 ，主要用于定期做一些事情
type Ticker struct {
C <-chan Time // the channel on which the ticks are delivered.
// contains filtered or unexported fields
...
}
定期就是时间间隔，时间间隔的单位是 ns（纳秒，int64），在工厂函数 time.NewTicker 中以 Duration 类型的参数传入：func NewTicker(dur) *Ticker
在协程周期性的执行一些事情（打印状态日志，输出，计算等等）的时候非常有用。调用 Stop() 使计时器停止，在 defer 语句中使用。
ticker := time.NewTicker(updateInterval)
defer ticker.Stop()
...
select {
case u:= <-ch1:
...
case v:= <-ch2:
...
case <-ticker.C: //C是一个通道，数据为时间
logState(status) // call some logging function logState
default: // no value ready to be received
...
}
time.Tick()
time.Tick() 函数声明为 Tick(d Duration) <-chan Time，当你想返回一个通道而不必关闭它的时候这个函数非常有用：它以 d 为周期给返回的通道发送时间，d 是纳秒数。通过通道的阻塞作用和time的时间性，可以限制处理频率（函数 client.Call() 是一个 RPC 调用，这里暂不赘述（参见第 15.9 节））
import "time"
rate_per_sec := 10
var dur Duration = 1e9 / rate_per_sec
chRate := time.Tick(dur) // a tick every 1/10th of a second
for req := range requests {
<- chRate // rate limit our Service.Method RPC calls
go client.Call("Service.Method", req, ...)
}
每隔一段时间发一个并发请求，处理的频率可以根据机器负载（和/或）资源的情况而增加或减少。
time.Timer()
定时器 (Timer) 结构体看上去和计时器 (Ticker) 结构体的确很像（构造为 NewTimer(d Duration)），但是它只发送一次时间，在 Dration d （一段时间）之后。
time.After()
func After(d Duration) <-chan Time
在 Duration d 之后，当前时间被发到返回的通道，After() 只发送一次时间
package main
import (
"fmt"
"time"
)
func main() {
tick := time.Tick(1e8)//毫秒
boom := time.After(5e8)
for {
select {
case <-tick:
fmt.Println("tick.")
case <-boom: //只会命中一次
fmt.Println("BOOM!")
return
default:
fmt.Println(" .")
time.Sleep(5e7)
}
}
}
习惯用法：简单超时模式
一个操作只让它尝试一定的时间，一定时间后放弃
timeout := make(chan bool, 1)
go func() {
time.Sleep(1e9) // one second
timeout <- true
}()
select {
case <-ch://ch一秒后无数据就放弃，执行case <-timeout
// a read from ch has occured
case <-timeout:
// the read from ch has timed out
break
}
习惯用法：取消耗时很长的同步调用
使用 time.After() 函数替换 timeout-channel。
可以在 select 中通过 time.After() 发送的超时信号来停止协程的执行。以下代码，在 timeoutNs 纳秒后执行 select 的 timeout 分支后，执行 client.Call 的协程也随之结束，不会给通道 ch 返回值：
注意:缓冲大小设置为 1 是必要的，可以避免协程死锁以及确保超时的通道可以被垃圾回收。（我的试验，并没有发生死锁，协程+通道 独立存在也不会发生死锁,会自己结束）
ch := make(chan error, 1)
go func() { ch <- client.Call("Service.Method", args, &reply) } ()
select {
case resp := <-ch
// use resp and reply
case <-time.After(timeoutNs):
// call timed out
break
}
试验代码：
package main
import (
"fmt"
"time"
)
func main() {
ch := make(chan int)
go func() { ch <- fn() }()
select {
case resp := <-ch:
fmt.Println(resp)
// use resp and reply
case <-time.After(9 * 1e9):
// call timed out
break
}
}
func fn() int {
time.Sleep(100 * 1e9)
return 1
}
此外，需要注意在有多个 case 符合条件时， select 对 case 的选择是伪随机的，如果上面的代码稍作修改如下，则 select 语句可能不会在定时器超时信号到来时立刻选中 time.After(timeoutNs) 对应的 case，因此协程可能不会严格按照定时器设置的时间结束。
ch := make(chan int, 1)
go func() { for { ch <- 1 } } ()
L:
for {
select {
case <-ch:// 这个分支总是符合条件
// do something
case <-time.After(timeoutNs):
// call timed out
break L
}
}
以下的用法就一般：感觉没必要用select
func Query(conns []Conn, query string) Result {
ch := make(chan Result, 1)
for _, conn := range conns {
go func(c Conn) {
select {
case ch <- c.DoQuery(query):
default:
}
}(conn)
}
return <- ch
}
再次声明，结果通道 ch 必须是带缓冲的：以保证第一个发送进来的数据有地方可以存放，确保放入的首个数据总会成功（而不是等到收发双方同步后再 放数据），这样，应用比较友好，不会请求过久，这就是缓冲的意义
正在执行的协程可以总是可以使用 runtime.Goexit() 来停止。
协程和恢复 (recover)
补充1：
若defer在panic()前，那么可以正常触发
package main
import (
"fmt"
)
func main() {
defer fmt.Println("defer")
fmt.Println("正常")
panic("抛出异常")
}
补充2：
正如名字一样，这个 (recover()) 内建函数被用于从 panic 或错误场景中恢复：让程序可以从 panicking 重新获得控制权，停止终止过程进而恢复正常执行。
recover 只能在 defer 修饰的函数（参见 6.4 节）中使用：用于取得 panic() 调用中传递过来的错误值；
如果是正常执行，调用 recover() 会返回 nil，且没有其它效果。
总结：panic() 会导致栈被展开直到 defer 修饰的 recover() 被调用或者程序中止。
下面例子中的 protect() 函数调用函数参数 g 来保护调用者防止从 g 中抛出的运行时 panic，并展示 panic 中的信息：
func protect(g func()) {
defer func() {
log.Println("done")
// Println executes normally even if there is a panic
if err := recover(); err != nil {
log.Printf("run time panic: %v", err)
}
}()
log.Println("start")
g() // possible runtime-error
}
这跟 Java 和 .NET 这样的语言中的 catch 块类似。
应用场景
log 包实现了简单的日志功能：默认的 log 对象向标准错误输出中写入并打印每条日志信息的日期和时间，如包中的Println、Printf。
除了 Println 和 Printf 函数，其它的致命性函数都会在写完日志信息后调用 os.Exit(1)，那些退出函数也是如此。
而 Panic 效果的函数会在写完日志信息后调用 panic()；
可以在程序必须中止或发生了临界错误时使用它们，就像当 web 服务器不能启动时那样（参见 15.4 节 中的例子）。
总结：在程序必须中止或发生了临界错误时，可以使用os.Exit(1)效果的函数，如log.Println() log.Println() ，也可以使用Panic效果的函数panic()
// panic_recover.go
package main
import (
"fmt"
)
func badCall() {
panic("bad end")
}
func test() {
defer func() {
if e := recover(); e != nil {
fmt.Printf("Panicing %s\r\n", e)
}
}()
badCall()
fmt.Printf("After bad call\r\n") // <-- would not reach 不会运行
}
func main() {
fmt.Printf("Calling test\r\n")
test()
fmt.Printf("Test completed\r\n") //虽然test()剩下的不保，但是test()后面的保住了，这行可以运行
}
输出
Calling test
Panicing bad end
Test completed
一个用到 recover() 的 协程 程序
func server(workChan <-chan *Work) {
for work := range workChan {
go safelyDo(work) // start the goroutine for that work
}
}
func safelyDo(work *Work) {
defer func() {
if err := recover(); err != nil {
log.Printf("Work failed with %s in %v", err, work)
}
}()
do(work)
}
上边的代码，如果 do(work) 发生 panic()，错误会被记录且协程会退出并释放，而其他协程不受影响。
新旧模型对比：任务和 worker
传统模式：共享内存
//任务
type Task struct {
// some state
}
//任务池
type Pool struct {
Mu sync.Mutex
Tasks []*Task
}
//工作者（处理者）
func Worker(pool *Pool) {
for {
pool.Mu.Lock()
// begin critical section:
task := pool.Tasks[0] // take the first task
pool.Tasks = pool.Tasks[1:] // update the pool of tasks
// end critical section
pool.Mu.Unlock()
process(task)
}
}
sync.Mutex（参见9.3）是互斥锁：它用来在代码中保护临界区资源：同一时间只有一个 go 协程 (goroutine) 可以进入该临界区。如果出现了同一时间多个 go 协程都进入了该临界区，则会产生竞争：Pool 结构就不能保证被正确更新。传统的模式，经典的面向对象的语言中应用得比较多，比如 C++，JAVA，C#
加锁保证了同一时间只有一个 go 协程可以进入到 pool 中：一项任务有且只能被赋予一个 worker 。
如果不加锁，则工作协程可能会在 task:=pool.Tasks[0] 发生切换，导致 pool.Tasks=pool.Tasks[1:] 结果异常：一些 worker 获取不到任务，而一些任务可能被多个 worker 得到。
加锁实现同步的方式在工作协程比较少时可以工作得很好，但是当工作协程数量很大，任务量也很多时，处理效率将会因为频繁的加锁/解锁开销而降低。当工作协程数增加到一个阈值时，程序效率会急剧下降，这就成为了瓶颈。
新模式：使用通道
使用通道进行同步：使用一个通道接受需要处理的任务，一个通道接受处理完成的任务（及其结果）。worker 在协程中启动，其数量 N 应该根据任务数量进行调整。
package main
import (
"fmt"
"time"
)
type Task struct {
Name string
}
type Pool struct {
Tasks []*Task
}
const N = 1
func main() {
pending, done := make(chan *Task), make(chan *Task)
tasks := []*Task{{Name: "t1"}, {Name: "t2"}, {Name: "t3"}, {Name: "t5"}}
pool := Pool{}
pool.Tasks = tasks
go sendWork(pending, pool)
for i := 0; i < N; i++ {
go Worker(pending, done)
}
go consumeWork(done)
time.Sleep(5 * 1e9) // 不写的话，主协程马上结束，其他协程无法执行
}
func sendWork(pending chan *Task, pool Pool) {
for i := 0; i < len(pool.Tasks); i++ {
v := pool.Tasks[i]
pending <- v
}
}
func Worker(in, out chan *Task) {
for t := range in { // 会自动检测in是否关闭
time.Sleep(1 * 1e9) //处理中
out <- t
}
}
func consumeWork(in chan *Task) {
for t := range in {
fmt.Println(t.Name + "已经处理好了")
}
}
（随着sendWork的结束，Worker和consumeWork会一直在阻塞，但是没关系，主协程没有阻塞，所以不会发生死锁）
随着任务数量增加，worker 数量也应该相应增加，同时性能并不会像第一种方式那样下降明显。在 pending 通道中存在一份任务的拷贝，第一个 worker 从 pending 通道中获得第一个任务并进行处理，这里并不存在竞争（对一个通道读数据和写数据的整个过程是原子性的：参见 14.2.2）。某一个任务会在哪一个 worker 中被执行是不可知的，反过来也是。worker 数量的增多也会增加通信的开销，这会对性能有轻微的影响。
从这个简单的例子中可能很难看出第二种模式的优势，但含有复杂锁运用的程序不仅在编写上显得困难，也不容易编写正确，使用第二种模式的话，就无需考虑这么复杂的东西了。
因此，第二种模式对比第一种模式而言，不仅性能是一个主要优势，而且还有个更大的优势：代码显得更清晰、更优雅。
应用
对于任何可以建模为 Master-Worker 范例的问题，一个类似于 worker 使用通道进行通信和交互、Master 进行整体协调的方案都能完美解决。
通道是一个较新的概念，本节我们着重强调了在 go 协程里通道的使用，但这并不意味着经典的锁方法就不能使用。go 语言让你可以根据实际问题进行选择：创建一个优雅、简单、可读性强、在大多数场景性能表现都能很好的方案。如果你的问题适合使用锁，也不要忌讳使用它。go 语言注重实用，什么方式最能解决你的问题就用什么方式，而不是强迫你使用一种编码风格。下面列出一个普遍的经验法则：
使用锁的情景：
访问共享数据结构中的缓存信息
保存应用程序上下文和状态信息数据
使用通道的情景：
与异步操作的结果进行交互
分发任务
传递数据所有权
总结
（我觉得单步用锁，多步用通道）
当你发现你的锁使用规则变得很复杂时，可以反省使用通道会不会使问题变得简单些。
惰性生成器的实现
生成器是指当被调用时返回一个序列中下一个值的函数
generateInteger() => 0
generateInteger() => 1
generateInteger() => 2
....
调一次生成一个，什么时候要，什么时候调，但是如果需要生成一个无限数量的偶数序列：一个一个调可能会很困难，而且内存会溢出！但是一个含有通道和 go 协程的函数能轻易实现这个需求。
resume (中断后)继续、恢复
yield 让步
package main
import (
"fmt"
)
var resume chan int
func integers() chan int {
yield := make(chan int)
count := 0
go func() {
for {
yield <- count
count++
}
}()
return yield
}
func generateInteger() int {
return <-resume
}
func main() {
resume = integers() //内部的通道yield移交给外部的通道resume
fmt.Println(generateInteger()) //=> 0
fmt.Println(generateInteger()) //=> 1
fmt.Println(generateInteger()) //=> 2
}
原来的，要想或得递增数列，必须有一个函数外变量，可以存储上一次的值，作为下一次的基础，然后多次调用 内存中就有多份函数同时运行。现在基于通道，就不必使用一个全局变量，内存中只有一份函数运行。（程序中的resume可以声明在main内）
有一个细微的区别是从通道读取的值可能会是稍早前产生的，并不是在程序被调用时生成的。
改进版：
改进版基于工厂模式，可以不改造自己的函数，而是 提供自己的函数和一个 初始状态（初始值），工厂函数会把你的函数改造成具有控制权的 生成函数，还是具有上面有点，单份函数运行。BuildLazyIntEvaluator和BuildLazyEvaluator都使用了闭包，保存了一些以后要用的东西。在协程中执行 无限循环，这个函数永远不会结束，并且永远可以通过通道控制它。
英语缩略词“RET”经常作为“Return from Procedure”的缩写来使用
package main
import (
"fmt"
)
type Any interface{}
type EvalFunc func(Any) (Any, Any)
func main() {
evenFunc := func(state Any) (Any, Any) { //惰性生成函数
os := state.(int)
ns := os + 2
return os, ns
}
even := BuildLazyIntEvaluator(evenFunc, 0)
for i := 0; i < 10; i++ {
fmt.Printf("%vth even: %v\n", i, even()) // 调用生成int控制器
}
}
func BuildLazyIntEvaluator(evalFunc EvalFunc, initState Any) func() int {
ef := BuildLazyEvaluator(evalFunc, initState)
return func() int {//
// 调用控制器，得到生成的 数
return ef().(int) //断言，如果转换合法，返回值 是 ef() 转换到 int 的值
}
}
func BuildLazyEvaluator(evalFunc EvalFunc, initState Any) func() Any {
retValChan := make(chan Any)
loopFunc := func() {
var actState Any = initState
var retVal Any
for {
retVal, actState = evalFunc(actState)
retValChan <- retVal
}
}
retFunc := func() Any { //上面 无限循环函数 的 控制器，基于通道控制
return <-retValChan
}
go loopFunc()
return retFunc
}
实现 Futures 模式
所谓 Futures 就是指：有时候在你使用某一个值之前需要先对其进行计算。这种情况下，你就可以在另一个处理器上进行该值的计算，到使用时，该值就已经计算完毕了。（体现的就是 并行 的思维）
原来的模式：
先等Inverse(a)完成，再等Inverse(b)完成
func InverseProduct(a Matrix, b Matrix) {
a_inv := Inverse(a)
b_inv := Inverse(b)
return Product(a_inv, b_inv)
}
现在的模式：
InverseFuture 和 InverseFuture 内部是协程，也就是不阻塞main协程，两者并发（并行），所以InverseFuture(a)不需要完成计算就马上执行InverseFuture(b)，两者都返回一个通道，只要监听两个通道都完成就可以马上进行下一步
func InverseProduct(a Matrix, b Matrix) {
a_inv_future := InverseFuture(a) // start as a goroutine
b_inv_future := InverseFuture(b) // start as a goroutine
a_inv := <-a_inv_future
b_inv := <-b_inv_future
return Product(a_inv, b_inv)
}
复用
典型的客户端/服务器（C/S）模式
使用 Go 的服务器通常会在协程中执行向客户端的响应，故而会对每一个客户端请求启动一个协程。一个常用的操作方法是客户端请求自身中包含一个通道，而服务器则向这个通道发送响应。
客户端：chan
服务器：chan <-
type Reply struct{...}
type Request struct{
arg1, arg2, arg3 some_type // 请求参数
replyc chan *Reply // 服务器响应的入口
}
type binOp func(a, b int) int //业务函数
func run(op binOp, req *Request) {
req.replyc <- op(req.a, req.b) //执行响应
}
server() 协程会无限循环以从 chan *Request 接收请求，并且为了避免被长时间操作所堵塞，它将为每一个请求启动一个协程来做具体的工作：
service（对顾客的）接待
func server(op binOp, service chan *Request) {
for {
req := <-service; // requests arrive here
// start goroutine for request:
go run(op, req); // don’t wait for op to complete
}
}
前面讲到，一个请求一个协程配对处理，所以server要以协程启动
返回值是chan *Request，因为前面说过，返回值时通过Request自带的通道给带回去的
func startServer(op binOp) chan *Request {
reqChan := make(chan *Request);
go server(op, reqChan);
return reqChan;
}
下面编写的检查响应
func main() {
adder := startServer(func(a, b int) int { return a + b })
const N = 100
var reqs [N]Request
for i := 0; i < N; i++ {
req := &reqs[i]
req.a = i
req.b = i + N
req.replyc = make(chan int)
adder <- req // adder is a channel of requests
}
// checks:
for i := N - 1; i >= 0; i-- {
// doesn’t matter what order
if <-reqs[i].replyc != N+2*i {
fmt.Println(“fail at”, i)
} else {
fmt.Println(“Request “, i, “is ok!”)
}
}
fmt.Println(“done”)
}
这个程序仅启动了 100 个协程。然而即使执行 100,000 个协程我们也能在数秒内看到它完成。这说明了 Go 的协程是如何的轻量：如果我们启动相同数量的真实的线程，程序早就崩溃了。（说明协程映射并分割线程，说明线程是能力过剩的）
完整代码
package main
import "fmt"
type Request struct {
a, b int
replyc chan int // reply channel inside the Request
}
type binOp func(a, b int) int
func run(op binOp, req *Request) {
req.replyc <- op(req.a, req.b)
}
func server(op binOp, service chan *Request) {
for {
req := <-service // requests arrive here
// start goroutine for request:
go run(op, req) // don't wait for op
}
}
func startServer(op binOp) chan *Request {
reqChan := make(chan *Request)
go server(op, reqChan)
return reqChan
}
func main() {
adder := startServer(func(a, b int) int { return a + b }) //参数业务处理函数
// 模拟请求
const N = 100
var reqs [N]Request
for i := 0; i < N; i++ {
req := &reqs[i] //引用结构体
req.a = i
req.b = i + N
req.replyc = make(chan int)
// 发生请求
adder <- req
}
// checks:检查响应情况
for i := N - 1; i >= 0; i-- { // doesn't matter what order
if <-reqs[i].replyc != N+2*i {
fmt.Println("fail at", i)
} else {
fmt.Println("Request ", i, " is ok!")
}
}
fmt.Println("done")
}
卸载 (Teardown)：通过信号通道关闭服务器
在上一个版本中 server() 在 main() 函数返回后并没有完全关闭，而被强制结束了(协程被主协程强制结束)。为了改进这一点，我们可以提供一个退出通道给 server() ：
func startServer(op binOp) (service chan *Request, quit chan bool) {
service = make(chan *Request)
quit = make(chan bool)
go server(op, service, quit)
return service, quit
}
server() 函数现在则使用 select 在 service 通道和 quit 通道之间做出选择：
func server(op binOp, service chan *request, quit chan bool) {
for {
select {
case req := <-service:
go run(op, req)
case <-quit:
return
}
}
}
完整代码：
完成请求后关闭服务
package main
import "fmt"
type Request struct {
a, b int
replyc chan int // reply channel inside the Request
}
type binOp func(a, b int) int
func run(op binOp, req *Request) {
req.replyc <- op(req.a, req.b)
}
func server(op binOp, service chan *Request, quit chan bool) {
for {
select {
case req := <-service:
go run(op, req)
case <-quit:
return
}
}
}
func startServer(op binOp) (service chan *Request, quit chan bool) {
service = make(chan *Request)
quit = make(chan bool)
go server(op, service, quit)
return service, quit
}
func main() {
adder, quit := startServer(func(a, b int) int { return a + b })
const N = 100
var reqs [N]Request
for i := 0; i < N; i++ {
req := &reqs[i]
req.a = i
req.b = i + N
req.replyc = make(chan int)
adder <- req
}
// checks:
for i := N - 1; i >= 0; i-- { // doesn't matter what order
if <-reqs[i].replyc != N+2*i {
fmt.Println("fail at", i)
} else {
fmt.Println("Request ", i, " is ok!")
}
}
quit <- true
fmt.Println("done")
}
限制同时处理的请求数
使用带缓冲区的通道很容易实现这一点（参见 14.2.5），其缓冲区容量就是同时处理请求的最大数量。程序 max_tasks.go 虽然没有做什么有用的事但是却包含了这个技巧：超过 MAXREQS 的请求将不会被同时处理，因为当信号通道表示缓冲区已满时 handle() 函数会阻塞且不再处理其他请求，直到某个请求从 sem 中被移除。sem 就像一个信号量，这一专业术语用于在程序中表示特定条件的标志变量。
（可用于开发 秒杀 业务）
package main
const MAXREQS = 50
var sem = make(chan int, MAXREQS)
type Request struct {
a, b int
replyc chan int
}
func process(r *Request) {
// do something
}
func handle(r *Request) {
sem <- 1 // doesn't matter what we put in it
process(r)
<-sem // one empty place in the buffer: the next request can start
}
func server(service chan *Request) {
for {
request := <-service
go handle(request)
}
}
func main() {
service := make(chan *Request)
go server(service)
}
通过这种方式，应用程序可以通过使用缓冲通道（通道被用作信号量）使协程同步其对该资源的使用，从而充分利用有限的资源（如内存）。
链式协程
当循环完成之后，一个 0 被写入到最右边的通道里，于是 100,000 个协程开始执行，接着 1000000 这个结果会在 1.5 秒之内被打印出来。
这个程序同时也展示了如何通过 flag.Int 来解析命令行中的参数以指定协程数量，例如：chaining -n=7000 会生成 7000 个协程。
package main
import (
"flag"
"fmt"
)
// 命令行参数叫做n，默认值100000，使用提示是how many goroutines
var ngoroutine = flag.Int("n", 100000, "how many goroutines")
func f(left, right chan int) { left <- 1 + <-right }
func main() {
flag.Parse()
leftmost := make(chan int)
var left, right chan int = nil, leftmost // 所谓的右，就是 最大的左
for i := 0; i < *ngoroutine; i++ { // *ngoroutine取出解析后的参数值
// 右边总是新的通道（int 为0），右边的通道总是 +1 给左边的通道，左边的通道
// 总是依赖右边的通道，所谓右边的通道，就是最大的左通道，两者是一个东西
left, right = right, make(chan int)
go f(left, right)
}
right <- 0 // bang!
x := <-leftmost // wait for completion
fmt.Println(x) // 100000, about 1.5 s
}
执行 go run . -n=5000
输出 5000
for 循环中最初的 go f(left, right) 因为没有发送者一直处于等待状态
在多核心上并行计算
假设我们有 NCPU 个 CPU 核心：const NCPU = 4 //对应一个四核处理器 然后我们想把计算量分成 NCPU 个部分，每一个部分都和其他部分并行运行。（NCPU如四核CPU）
总结：说白了，就是把协程数量控制在核心数以内 配合 runtime.GOMAXPROCS(NCPU) 开启多核心处理模式
func DoAll(){
sem := make(chan int, NCPU) // Buffering optional but sensible
for i := 0; i < NCPU; i++ {
go DoPart(sem)
}
// Drain the channel sem, waiting for NCPU tasks to complete
for i := 0; i < NCPU; i++ {
<-sem // wait for one task to complete
}
// All done.
}
func DoPart(sem chan int) {
// do the part of the computation
sem <-1 // signal that this piece is done
}
func main() {
runtime.GOMAXPROCS(NCPU) // runtime.GOMAXPROCS = NCPU
DoAll()
}
DoAll() 函数创建了一个 sem 通道，每个并行计算都将在对其发送完成信号；在一个 for 循环中 NCPU 个协程被启动了，每个协程会承担 1/NCPU 的工作量。每一个 DoPart() 协程都会向 sem 通道发送完成信号。
DoAll() 会在 for 循环中等待 NCPU 个协程完成：sem 通道就像一个信号量，这份代码展示了一个经典的信号量模式。（参见 14.2.7）
在以上运行模型中，您还需将 GOMAXPROCS 设置为 NCPU（参见 14.1.3）
并行化大量数据的计算
假设我们需要处理一些数量巨大且互不相关的数据项，它们从一个 in 通道被传递进来，当我们处理完以后又要将它们放入另一个 out 通道，就像一个工厂流水线一样。处理每个数据项也可能包含许多步骤：
Preprocess（预处理） / StepA（步骤A） / StepB（步骤B） / ... / PostProcess（后处理）
常规写法：
func SerialProcessData(in <-chan *Data, out chan<- *Data) {
for data := range in { // 无限循环
tmpA := PreprocessData(data)
tmpB := ProcessStepA(tmpA)
tmpC := ProcessStepB(tmpB)
out <- PostProcessData(tmpC)
}
}
通道写法：
func ParallelProcessData (in <-chan *Data, out chan<- *Data) {
// make channels:
preOut := make(chan *Data, 100)
stepAOut := make(chan *Data, 100)
stepBOut := make(chan *Data, 100)
stepCOut := make(chan *Data, 100)
// start parallel computations:
go PreprocessData(in, preOut)
go ProcessStepA(preOut,StepAOut)
go ProcessStepB(StepAOut,StepBOut)
go ProcessStepC(StepBOut,StepCOut)
go PostProcessData(StepCOut,out)
}
我认为区别就是，把按变量传递改为按通道传递，比如B在A之后，用到A的结果，虽然A与B不可同时进行，但是大量的A可以同时进行，然后一次性按批给B，这样大量的B又可以同时进行，而不是AB，再AB，再AB，但是单次一个运行
漏桶算法
https://cloud.tencent.com/developer/article/1711670
漏桶算法是一种限流算法：就是一个桶，底部有个洞，无论桶的水满了、溢出了还是没水，底部漏出的水都是一样的速率，除非没水了，出水的速率才改变，变为0/s。
在大量并发的环境下，为了防止由于请求暴涨，导致系统崩溃从而引起雪崩，一般会对流量做一定的限制操作。比如等待、排队、降级、拒绝服务、限流等。说明
漏桶算法(Leaky Bucket)是网络世界中流量整形或速率限制（Rate Limiting）时经常使用的一种算法，它的主要目的是控制数据注入到网络的速率，平滑网络上的突发流量。漏桶算法提供了一种机制，通过它，突发流量可以被整形以便为网络提供一个稳定的流量。
漏斗有一个入水口，一个出水口，出水口按照一定的速率出水，并且有一个最大出水速率。
1.入水速率小于等于出水速率的时候，漏斗内不会积水；
2.入水速率大于出水速率的时候，漏斗内会存在积水。
在漏斗内有水的情况下：
出水口按照最大速率出水；
漏斗未满的情况下，多出来的水会存在漏斗中；
漏斗满了的话，还有水进入漏斗，水会溢出。
基本过程
到达的数据包（网络层的PDU）被放置在底部具有漏孔的桶中（数据包缓存）
漏桶最多可以排队b个字节，漏桶的这个尺寸受限于有效的系统内存。如果数据包漏桶已经满了，那么数据包应被丢弃
数据包从漏桶中漏出，以常量速率（r字节/秒）注入网络，因此平滑了突发流量
漏桶算法强调的是在桶中缓存数据包，然后以一定的速率从桶中取出数据包，从而实现了限流，防止突发流量
总结：设置处理能力（漏桶的缓冲大小），不限制进水，但是超出处理能力的部分水舍弃，出水频率恒定，这就是以控制进水来控制出水
链接：https://www.jianshu.com/p/301727025990
allocate 分配
上面算法的实现，考虑以下的客户端-服务器结构：
客户端协程执行一个无限循环从某个源头（也许是网络）接收数据；数据读取到 Buffer 类型的缓冲区（把Buffer当成一个个数据）。为了避免分配过多的缓冲区以及释放缓冲区（缓冲区不能开太多），它保留了一份空闲缓冲区列表，并且使用一个缓冲通道来表示这个列表：var freeList = make(chan *Buffer,100)
这个可重用的缓冲区队列 (freeList) 与服务器是共享的。 当接收数据时，客户端尝试从 freeList 获取缓冲区；但如果此时通道为空，则会分配新的缓冲区。一旦消息被加载后，它将被发送到服务器上的 serverChan 通道：
var serverChan = make(chan *Buffer)
以下实现代码不必过分纠结！！
以下是客户端的算法代码：
func client() {
for {
var b *Buffer
// Grab a buffer if available; allocate if not
select {
case b = <-freeList:
// Got one; nothing more to do
default:
// None free, so allocate a new one
b = new(Buffer)
}
loadInto(b) // 把其他地方（如网络）的数据放进缓冲池
serverChan <- b // Send to server
}
}
服务器的循环则接收每一条来自客户端的消息并处理它，之后尝试将缓冲返回给共享的空闲缓冲区：
func server() {
for {
b := <-serverChan // Wait for work.
process(b)
// Reuse buffer if there's room.
select {
case freeList <- b:
// Reuse buffer if free slot on freeList; nothing more to do
default:
// Free list full, just carry on: the buffer is 'dropped'
}
}
}
但是这种方法在 freeList 通道已满的时候是行不通的，因为无法放入空闲 freeList 通道的缓冲区会被“丢到地上”由垃圾收集器回收（故名：漏桶算法）。
我的理解：
一开始，来1000条请求，可重用的缓冲区队列（freeList）为0，全部走default生成，服务器处理1000条，前100条丢进 freeList，后900全部丢弃。每次仅能漏出100条完成的请求，故称漏桶。又来1000条请求，前100条走freeList（此时可重用为100），余下900走default生成，在服务器处理1000条，前100前100条丢进 freeList（因为可重用已经清空），后900全部丢弃。
令牌桶
https://cloud.tencent.com/developer/article/1711670
令牌桶这种控制机制基于令牌桶中是否存在令牌来指示什么时候可以发送流量。令牌桶中的每一个令牌都代表一个字节(对于流量整形来说代表一个bit，就traffic policing来讲代表一个byte。）。如果令牌桶中存在令牌，则允许发送流量；而如果令牌桶中不存在令牌，则不允许发送流量。因此，如果突发门限被合理地配置并且令牌桶中有足够的令牌，那么流量就可以以峰值速率发送。
1 令牌桶有大小限制，不能无限放入令牌
2 令牌产生速率相同，如果令牌桶被耗尽，则整个处理速度则被令牌产生速度限制
3 令牌桶的令牌消耗速度不限，可以快速消耗完，也可以慢慢消耗
漏桶算法、令牌桶两种算法的区别
这两种算法的主要区别在于“漏桶算法”能够强行限制数据的传输速率，而“令牌桶算法”在能够限制数据的平均传输数据外，还允许某种程度的突发传输。在“令牌桶算法”中，只要令牌桶中存在令牌，那么就允许突发地传输数据直到达到用户配置的门限，因此它适合于具有突发特性的流量。
对 Go 协程进行基准测试
基准测试只有在所有的测试通过后才能运行！
此我们将其应用到一个用协程向通道写入整数再读出的实例中。这个函数将通过 testing.Benchmark() 调用 N 次（例如：N = 1,000,000），BenchMarkResult 有一个 String() 方法来输出其结果。N 的值将由 gotest 来判断并取得一个足够大的数字，以获得合理的基准测试结果。
如果你想排除指定部分的代码或者更具体的指定要测试的部分，可以使用 testing.B.startTimer() 和 testing.B.stopTimer() 来开始或结束计时器。
mark 打分
package main
import (
"fmt"
"testing"
)
func main() {
fmt.Println(" sync", testing.Benchmark(BenchmarkChannelSync).String())
fmt.Println("buffered", testing.Benchmark(BenchmarkChannelBuffered).String())
}
func BenchmarkChannelSync(b *testing.B) {
ch := make(chan int)
go func() {
for i := 0; i < b.N; i++ {
ch <- i
}
close(ch)
}()
for range ch {
}
}
func BenchmarkChannelBuffered(b *testing.B) {
ch := make(chan int, 128)
go func() {
for i := 0; i < b.N; i++ {
ch <- i
}
close(ch)
}()
for range ch {
}
}
sync 4737838 253.6 ns/op
buffered 17143297 65.51 ns/op
以后用到再说
使用通道并发访问对象
为了保护对象被并发访问修改，我们可以使用协程在后台顺序执行匿名函数来替代使用同步互斥锁。（前面讲了很多次，其实就是PV操作代替锁）
package main
import (
"fmt"
"strconv"
)
type Person struct {
Name string
salary float64
chF chan func()
}
func NewPerson(name string, salary float64) *Person {
p := &Person{name, salary, make(chan func())}
go p.backend()
return p
}
func (p *Person) backend() {
for f := range p.chF { //无限循环
f()
}
}
// Set salary.
func (p *Person) SetSalary(sal float64) {
p.chF <- func() { p.salary = sal }
}
// Retrieve salary.
func (p *Person) Salary() float64 {
fChan := make(chan float64)
p.chF <- func() { fChan <- p.salary }
return <-fChan
}
func (p *Person) String() string {
return "Person - name is: " + p.Name + " - salary is: " + strconv.FormatFloat(p.Salary(), 'f', 2, 64)
}
func main() {
// 单例模式，创建的person是独一份的，只在后台backend()不断循环检测信号（通道）
// 要想改值，必须先调用通道，这样保证相同时间内只有一个地方在访问salary，防止出现
//丢失更新，不可重复读，脏读
bs := NewPerson("Smith Bill", 2500.5)
fmt.Println(bs)
bs.SetSalary(4000.25) //解除通道阻塞并赋值
fmt.Println("Salary changed:")
fmt.Println(bs)
fmt.Println(bs.String())
}