协程与通道

什么是协程

一个应用程序是运行在机器上的一个进程；进程是一个运行在自己内存地址空间里的独立执行体。一个进程由一个或多个操作系统线程组成，这些线程其实是共享同一个内存地址空间的一起工作的执行体。

并行是一种通过使用多处理器以提高速度的能力。所以并发程序可以是并行的，也可以不是。

公认的，使用多线程的应用难以做到准确，最主要的问题是内存中的数据共享，它们会被多线程以无法预知的方式进行操作，导致一些无法重现或者随机的结果（称作 竞态）。

不要使用全局变量或者共享内存，它们会给你的代码在并发运算的时候带来危险。

在 Go 中，应用程序并发处理的部分被称作 goroutines（协程），它可以进行更有效的并发运算。在协程和操作系统线程之间并无一对一的关系：协程是根据一个或多个线程的可用性，映射（多路复用，执行于）在他们之上的；协程调度器在 Go 运行时很好的完成了这个工作。

协程是轻量的，比线程更轻。它们痕迹非常不明显（使用少量的内存和资源）：使用 4K 的栈内存就可以在堆中创建它们。因为创建非常廉价，必要的时候可以轻松创建并运行大量的协程（在同一个地址空间中 100,000 个连续的协程）。并且它们对栈进行了分割，从而动态的增加（或缩减）内存的使用；栈的管理是自动的，但不是由垃圾回收器管理的，而是在协程退出后自动释放。

存在两种并发方式：

• 确定性的（明确定义排序）

• 非确定性的（加锁 / 互斥从而未定义排序）。

Go 的协程和通道理所当然的支持确定性的并发方式（例如通道具有一个 sender 和一个 receiver）。

并发和并行的差异

Go 的并发原语提供了良好的并发设计基础：表达程序结构以便表示独立地执行的动作；所以 Go 的重点不在于并行的首要位置：并发程序可能是并行的，也可能不是。并行是一种通过使用多处理器以提高速度的能力。但往往是，一个设计良好的并发程序在并行方面的表现也非常出色。

使用 GOMAXPROCS

在 gc 编译器下（6g 或者 8g）你必须设置 GOMAXPROCS 为一个大于默认值 1 的数值来允许运行时支持使用多于 1 个的操作系统线程，否则所有的协程都会共享同一个线程。 当 GOMAXPROCS 大于 1 时，会有一个线程池管理众多线程。gccgo 编译器 会使 GOMAXPROCS 与运行中的协程数量相等。假设一个机器上有 n 个处理器或者核心。如果你设置环境变量 GOMAXPROCS>=n，或者执行 runtime.GOMAXPROCS(n)，那么协程会被分割（或分散）到 n 个处理器上。但是增加处理器数量并不意味着性能的线性提升。通常，如果有 n 个核心，会设置 GOMAXPROCS 为 n-1 以获得最佳性能，但同样也需要保证，协程的数量 > 1 + GOMAXPROCS > 1。

所以如果在某一时间只有一个协程在执行，不要设置 GOMAXPROCS！

如何用命令行指定使用的核心数量

使用 flags 包，如下：

var numCores = flag.Int("n", 2, "number of CPU cores to use")
​
in main()
flag.Parse()
runtime.GOMAXPROCS(*numCores)

协程可以通过调用 runtime.Goexit() 来停止，尽管这样做几乎没有必要。

Go 协程（goroutines）和协程（coroutines）

• Go 协程意味着并发（或者可以以并行的方式部署），协程一般来说不是这样的

• Go 协程通过通道来通信；协程通过让出和恢复操作来通信

Go 程（goroutine）是由 Go 运行时管理的轻量级线程。

go f(x, y, z)

会启动一个新的 Go 程并执行

f(x, y, z)

f, x, y 和 z 的求值发生在当前的 Go 程中，而 f 的执行发生在新的 Go 程中

package main
​
import ("fmt""time"
)
​
func say(s string) {for i := 0; i < 5; i++ {time.Sleep(100 * time.Millisecond)fmt.Println(s)}
}
​
func main() {go say("world")say("hello")
}

协程间的信道

概念

而 Go 有一个特殊的类型，通道（channel），像是通道（管道），可以通过它们发送类型化的数据在协程之间通信，可以避开所有内存共享导致的坑；通道的通信方式保证了同步性。数据通过通道：同一时间只有一个协程可以访问数据：所以不会出现数据竞争，设计如此。数据的归属（可以读写数据的能力）被传递。

通常使用这样的格式来声明通道：var identifier chan datatype

未初始化的通道的值是 nil。

所以通道只能传输一种类型的数据，比如 chan int 或者 chan string，所有的类型都可以用于通道，空接口 interface{} 也可以。

var ch1 chan string
ch1 = make(chan string)

当然可以更短： ch1 := make(chan string)。

通信操作符 <-

这个操作符直观的标示了数据的传输：信息按照箭头的方向流动。

• 流向通道（发送）

  • ch <- int1 表示：用通道 ch 发送变量 int1（双目运算符，中缀 = 发送）

• 从通道流出（接收），三种方式：

  • int2 = <- ch 表示：变量 int2 从通道 ch（一元运算的前缀操作符，前缀 = 接收）接收数据（获取新值）

  • 假设 int2 已经声明过了，如果没有的话可以写成：int2 := <- ch。

  • <- ch 可以单独调用获取通道的（下一个）值，当前值会被丢弃，但是可以用来验证，所以以下代码是合法的：

    if <- ch != 1000{...
    }

package main
​
import ("fmt""time"
)
​
func main() {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 <- "Tokio"
}
​
func getData(ch chan string) {var input string// time.Sleep(2e9)for {input = <-chfmt.Printf("%s ", input)}
}

通道阻塞

默认情况下，通信是同步且无缓冲的：在有接收者接收数据之前，发送不会结束。可以想象一个无缓冲的通道在没有空间来保存数据的时候：必须要一个接收者准备好接收通道的数据然后发送者可以直接把数据发送给接收者。所以通道的发送 / 接收操作在对方准备好之前是阻塞的：

• 对于同一个通道，发送操作（协程或者函数中的），在接收者准备好之前是阻塞的：如果 ch 中的数据无人接收，就无法再给通道传入其他数据：新的输入无法在通道非空的情况下传入。所以发送操作会等待 ch 再次变为可用状态：就是通道值被接收时（可以传入变量）。

• 对于同一个通道，接收操作是阻塞的（协程或函数中的），直到发送者可用：如果通道中没有数据，接收者就阻塞了。

package main
​
import ("fmt""time"
)
​
func main() {ch1 := make(chan int)go pump(ch1)go suck(ch1)time.Sleep(1e9)
}
​
func suck(ch chan int) {for {fmt.Println(<-ch)}
}
​
func pump(ch chan int) {for i := 0; ; i++ {ch <- i}
}

上面这段程序创建两个协程，一个用于发送一个用于接收，从开始运行直到 time.Sleep(1e9)代码运行完毕，程序结束。

通过一个（或多个）通道交换数据进行协程同步

通信是一种同步形式：通过通道，两个协程在通信（协程会和）中某刻同步交换数据。无缓冲通道成为了多个协程同步的完美工具。

发送者可通过 close 关闭一个信道来表示没有需要发送的值了。接收者可以通过为接收表达式分配第二个参数来测试信道是否被关闭：若没有值可以接收且信道已被关闭，那么在执行完

v, ok := <-ch

之后 ok 会被设置为 false。

循环 for i := range c 会不断从信道接收值，直到它被关闭。

注意：

• 只有发送者才能关闭信道，而接收者不能。向一个已经关闭的信道发送数据会引发程序恐慌（panic）。

• 信道与文件不同，通常情况下无需关闭它们。只有在必须告诉接收者不再有需要发送的值时才有必要关闭，例如终止一个 range 循环。

package main
​
import ("fmt"
)
​
func main() {a := 0c := make(chan int, 10)go fibonacci(cap(c), c)for i := range c {a++fmt.Println(i)}println(a)
}
​
func fibonacci(n int, c chan int) {x, y := 0, 1for i := 0; i < n; i++ {c <- xx, y = y, x+y}close(c)
}

死锁：

package main
​
import ("fmt"
)
​
func f1(in chan int) {fmt.Println(<-in)
}
​
func main() {out := make(chan int)out <- 2go f1(out)
}

同步通道 - 使用带缓冲的通道

一个无缓冲通道只能包含 1 个元素，有时显得很局限。我们给通道提供了一个缓存，可以在扩展的 make 命令中设置它的容量，如下：

buf := 100 ch1 := make(chan string, buf) buf 是通道可以同时容纳的元素（这里是 string）个数

在缓冲满载（缓冲被全部使用）之前，给一个带缓冲的通道发送数据是不会阻塞的，而从通道读取数据也不会阻塞，直到缓冲空了。

同步：ch :=make(chan type, value)

• value == 0 -> synchronous, unbuffered (阻塞）

• value > 0 -> asynchronous, buffered（非阻塞）取决于 value 元素

协程中用通道输出结果

信号量模式

使用通道让 main 程序等待协程完成

协程通过在通道 ch 中放置一个值来处理结束的信号。main 协程等待 <-ch 直到从中获取到值。

select 语句

select 语句使一个 Go 程可以等待多个通信操作。

select 会阻塞到某个分支可以继续执行为止，这时就会执行该分支。当多个分支都准备好时会随机选择一个执行。

从不同的并发执行的协程中获取值可以通过关键字 select 来完成，它和 switch 控制语句非常相似也被称作通信开关；它的行为像是 “你准备好了吗” 的轮询机制；select 监听进入通道的数据，也可以是用通道发送值的时候。（蛮像 juc 里面 nio 的 selector 选择器）

格式：

select {
case u:= <- ch1:...
case v:= <- ch2:......
default: // no value ready to be received...
}

例子：

package main
​
import ("fmt"
)
​
func main() {c := make(chan int, 10)quit := make(chan int)go func() {for i := 0; i < 10; i++ {fmt.Println(<-c)}quit <- 0}()fibonacci(c, quit)
}
​
func fibonacci(c, quit chan int) {x, y := 0, 1for {select {case c <- x:x, y = y, x+ycase <-quit:fmt.Println("quit")return}}
}

select 做的就是：选择处理列出的多个通信情况中的一个。

• 如果都阻塞了，会等待直到其中一个可以处理

• 如果多个可以处理，随机选择一个

• 如果没有通道操作可以处理并且写了 default 语句，它就会执行：default 永远是可运行的（这就是准备好了，可以执行）。

select 语句实现了一种监听模式，通常用在（无限）循环中；在某种情况下，通过 break 语句使循环退出。

默认选择

当 select 中的其它分支都没有准备好时，default 分支就会执行。

为了在尝试发送或者接收时不发生阻塞，可使用 default 分支：

select {
case i := <-c:// 使用 i
default:// 从 c 中接收会阻塞时执行
}

举例：

package main
​
import ("fmt""time"
)
​
func main() {tick := time.Tick(100 * time.Millisecond)boom := time.After(500 * time.Millisecond)for {select {case <-tick:fmt.Println("tick.")case <-boom:fmt.Println("BOOM!")returndefault:fmt.Println("    .")time.Sleep(50 * time.Millisecond)}}
}

sync.Mutex

Go 标准库中提供了 sync.Mutex 互斥锁类型及其两个方法：

• Lock

• Unlock

我们可以通过在代码前调用 Lock 方法，在代码后调用 Unlock 方法来保证一段代码的互斥执行。

package main
​
import ("fmt""sync""time"
)
​
// SafeCounter 的并发使用是安全的。
type SafeCounter struct {v   map[string]intmux sync.Mutex
}
​
// Inc 增加给定 key 的计数器的值。
func (c *SafeCounter) Inc(key string) {c.mux.Lock()// Lock 之后同一时刻只有一个 goroutine 能访问 c.vc.v[key]++c.mux.Unlock()
}
​
// Value 返回给定 key 的计数器的当前值。
func (c *SafeCounter) Value(key string) int {c.mux.Lock()// Lock 之后同一时刻只有一个 goroutine 能访问 c.vdefer c.mux.Unlock()return c.v[key]
}
​
func main() {c := SafeCounter{v: make(map[string]int)}for i := 0; i < 1000; i++ {go c.Inc("somekey")}
​time.Sleep(time.Second)fmt.Println(c.Value("somekey"))
}

协程和恢复（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)
}