时间戳

首页 > 技术文章 > A tour of Go (5) - 并发

A tour of Go (5) - 并发

lsc2019 2020-12-31 08:38 原文

Wednesday, December 30, 2020

A tour of Go (5) - 并发

  1. Go 程 (goroutine)
    Go 语言支持并发，我们只需要通过 go 关键字来开启 goroutine 即可。

goroutine 是轻量级线程，goroutine 的调度是由 Golang 运行时进行管理的。

go f(x, y, z) // 语法格式：go 函数名( 参数列表 )
会启动一个新的 Go 程并执行

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

同一个程序中的所有Go 程在相同的地址空间中运行，因此在访问共享的内存时必须进行同步。sync 包提供了这种能力，不过在 Go 中并不经常用到，因为还有其它的办法（见下一页）。

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")
}
/* 输出的 hello 和 world 是没有固定先后顺序。因为它们是两个 goroutine 在执行：
Results:
world
hello
world
hello
hello
world
hello
world
world
hello
*/
2. 信道/通道（channel）
通道（channel）是用来传递数据的一个数据结构。

通道可用于两个 goroutine 之间通过传递一个指定类型的值来同步运行和通讯。

信道是带有类型的管道，你可以通过它用信道操作符 <- 来发送或者接收值。如果未指定方向，则为双向通道。

ch <- v // 将 v 发送至信道 ch。
v := <-ch // 从 ch 接收值并赋予 v。
（“箭头”就是数据流的方向。）

和映射与切片一样，信道在使用前必须创建：

：
ch := make(chan int) // 声明一个通道：使用chan关键字
注意：默认情况下，通道是不带缓冲区的，发送和接收操作在另一端准备好之前都会阻塞。发送端发送数据，同时必须有接收端相应的接收数据。这使得 Go 程可以在没有显式的锁或竞态变量的情况下进行同步。

以下示例对切片中的数进行求和，将任务分配给两个 Go 程。一旦两个 Go 程完成了它们的计算，它就能算出最终的结果。

package main
​
import "fmt"
​
func sum(s []int, c chan int) {
sum := 0
for _, v := range s {
sum += v
}
c <- sum // 把 sum 发送到通道 c
}
​
func main() {
s := []int{7, 2, 8, -9, 4, 0}
​
c := make(chan int)
go sum(s[:len(s)/2], c)
go sum(s[len(s)/2:], c)
x, y := <-c, <-c // 从通道 c 中接收
​
fmt.Println(x, y, x+y) // Result:-5 17 12
}
带缓冲的信道：信道可以是 带缓冲的。将缓冲长度作为第二个参数提供给 make 来初始化一个带缓冲的信道：

ch := make(chan int, 100)
带缓冲区的通道允许发送端的数据发送和接收端的数据获取处于异步状态，就是说发送端发送的数据可以放在缓冲区里面，可以等待接收端去获取数据，而不是立刻需要接收端去获取数据。

仅当信道的缓冲区填满后，向其发送数据时才会阻塞。当缓冲区为空时，接受方会阻塞。

ch := make(chan int, 2)
ch <- 1
ch <- 2
fmt.Println(<-ch) // Result：1
fmt.Println(<-ch) // Result：2
// 如果填满缓冲区后继续发送，报错：fatal error: all goroutines are asleep - deadlock!
range和close：发送者可通过 close 关闭一个信道来表示没有需要发送的值了。接收者可以通过为接收表达式分配第二个参数来测试信道是否被关闭：若没有值可以接收且信道已被关闭，那么在执行完

v, ok := <-ch
之后 ok 会被设置为 false。

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

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

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

func fibonacci(n int, c chan int) { // n 为 channel c 的缓冲区长度
x, y := 0, 1 // 初始化 x, y 为第0位，第1位
for i := 0; i < n; i++ {
c <- x // 将 x 送入 c
x, y = y, x+y // x, y 一起后移
}
close(c) //
}
​
func main() {
c := make(chan int, 10)
//fmt.Println(cap(c))
go fibonacci(cap(c), c) // 将channel缓冲区长度传入函数
// range 函数遍历每个从通道接收到的数据，因为 c 在发送完 10 个
// 数据之后就关闭了通道，所以这里我们 range 函数在接收到 10 个数据
// 之后就结束了。如果上面的 c 通道不关闭，那么 range 函数就不
// 会结束，从而在接收第 11 个数据的时候就阻塞了。
for i := range c {
fmt.Println(i) // 按顺序print channel c 中的值
}
}
select 语句：select 语句使一个 Go 程可以等待多个通信操作。

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

/
package main
​
import "fmt"
​
func fibonacci(c, quit chan int) {
x, y := 0, 1
for {
select {
case c <- x:
x, y = y, x+y
case <-quit:
fmt.Println("quit")
return
}
}
}
​
func main() {
c := make(chan int) // 没有缓冲区
quit := make(chan int) // 没有缓冲区
go func() {
for i := 0; i < 10; i++ {
fmt.Println(<-c)
}
quit <- 0
}()
fibonacci(c, quit)
}
/
Resutls:
0
1
1
2
3
5
8
13
21
34
quit
*/
默认选择：当 select 中的其它分支都没有准备好时，default 分支就会执行。

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

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

tick := time.Tick(100 * time.Millisecond)
boom := time.After(500 * time.Millisecond)
for {
select {
case <-tick:
fmt.Println("tick.")
case <-boom:
fmt.Println("BOOM!")
return
default:
fmt.Println(" .")
time.Sleep(50 * time.Millisecond)
}
}
/*
Resutls:
.
.
tick.
.
.
tick.
.
.
tick.
.
.
tick.
.
.
tick.
BOOM!
*/
练习：等价二叉查找树 https://tour.go-zh.org/concurrency/7

题目描述：不同二叉树的叶节点上可以保存相同的值序列。例如，以下两个二叉树都保存了序列 1，1，2，3，5，8，13。

在大多数语言中，检查两个二叉树是否保存了相同序列的函数都相当复杂。 我们将使用 Go 的并发和信道来编写一个简单的解法。

本例使用了 tree 包，它定义了类型：

type Tree struct {
Left *Tree
Value int
Right *Tree
}

  1. 实现 Walk 函数。

  2. 测试 Walk 函数。

函数 tree.New(k) 用于构造一个随机结构的已排序二叉查找树，它保存了值 k, 2k, 3k, ..., 10k。

创建一个新的信道 ch 并且对其进行步进：

go Walk(tree.New(1), ch)
然后从信道中读取并打印 10 个值。应当是数字 1, 2, 3, ..., 10。

  1. 用 Walk 实现 Same 函数来检测 t1 和 t2 是否存储了相同的值。

  2. 测试 Same 函数。

Same(tree.New(1), tree.New(1)) 应当返回 true，而 Same(tree.New(1), tree.New(2)) 应当返回 false。

Tree 的文档可在这里找到。

sync.Mutex

我们已经看到信道非常适合在各个 Go 程间进行通信。

但是如果我们并不需要通信呢？比如说，若我们只是想保证每次只有一个 Go 程能够访问一个共享的变量，从而避免冲突？

这里涉及的概念叫做 互斥（mutualexclusion）* ，我们通常使用 互斥锁（Mutex） 这一数据结构来提供这种机制。

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

Lock

Unlock

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

我们也可以用 defer 语句来保证互斥锁一定会被解锁。参见 Value 方法。

package main
​
import (
"fmt"
"sync"
"time"
)
​
// SafeCounter 的并发使用是安全的。
type SafeCounter struct {
v map[string]int
mux sync.Mutex
}
​
// Inc 增加给定 key 的计数器的值。
func (c *SafeCounter) Inc(key string) {
c.mux.Lock()
// Lock 之后同一时刻只有一个 goroutine 能访问 c.v
c.v[key]++
c.mux.Unlock()
}
​
// Value 返回给定 key 的计数器的当前值。
func (c *SafeCounter) Value(key string) int {
c.mux.Lock()
// Lock 之后同一时刻只有一个 goroutine 能访问 c.v
defer 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"))
}
练习：Web爬虫 https://tour.go-zh.org/concurrency/10

在这个练习中，我们将会使用 Go 的并发特性来并行化一个 Web 爬虫。

修改 Crawl 函数来并行地抓取 URL，并且保证不重复。

提示：你可以用一个 map 来缓存已经获取的 URL，但是要注意 map 本身并不是并发安全的！

package main
​
import (
"fmt"
"sync"
)
​
type Fetcher interface {
// Fetch 返回 URL 的 body 内容，并且将在这个页面上找到的 URL 放到一个 slice 中。
Fetch(url string) (body string, urls []string, err error)
}
​
// Crawl 使用 fetcher 从某个 URL 开始递归的爬取页面，直到达到最大深度。
func Crawl(url string, depth int, fetcher Fetcher, crawled Crawled, out chan string, end chan bool) {
// TODO: 并行的抓取 URL。
// TODO: 不重复抓取页面。
// 下面并没有实现上面两种情况：
if depth <= 0 {
end <- true
return
}
​
crawled.mux.Lock()
if _, ok := crawled.crawled[url]; ok {
crawled.mux.Unlock()
end <- true
return
}
​
crawled.crawled[url] = 1
crawled.mux.Unlock()
​
_, urls, err := fetcher.Fetch(url)
if err != nil {
fmt.Println(err)
end <- true
return
}
​
out <- url
//fmt.Println("found: ", url, body)
for _, u := range urls {
go Crawl(u, depth-1, fetcher, crawled, out, end)
}
​
for i := 0; i < len(urls); i++ {
<-end
}
​
end <- true
return
}
​
type Crawled struct {
crawled map[string]int
mux sync.Mutex
}
func main() {
crawled := Crawled{make(map[string]int), sync.Mutex{}}
out := make(chan string)
end := make(chan bool)
go Crawl("http://golang.org/", 4, fetcher, crawled, out, end)
​
for {
select {
case url := <-out:
fmt.Println("found: ", url)
case <-end:
return
}
}
}
​
// fakeFetcher 是返回若干结果的 Fetcher。
type fakeFetcher map[string]*fakeResult
​
type fakeResult struct {
body string
urls []string
}
​
func (f fakeFetcher) Fetch(url string) (string, []string, error) {
if res, ok := f[url]; ok {
return res.body, res.urls, nil
}
return "", nil, fmt.Errorf("not found: %s", url)
}
​
// fetcher 是填充后的 fakeFetcher。
var fetcher = fakeFetcher{
"http://golang.org/": &fakeResult{
"The Go Programming Language",
[]string{
"http://golang.org/pkg/",
"http://golang.org/cmd/",
},
},
"http://golang.org/pkg/": &fakeResult{
"Packages",
[]string{
"http://golang.org/",
"http://golang.org/cmd/",
"http://golang.org/pkg/fmt/",
"http://golang.org/pkg/os/",
},
},
"http://golang.org/pkg/fmt/": &fakeResult{
"Package fmt",
[]string{
"http://golang.org/",
"http://golang.org/pkg/",
},
},
"http://golang.org/pkg/os/": &fakeResult{
"Package os",
[]string{
"http://golang.org/",
"http://golang.org/pkg/",
},
},
}

推荐阅读