golang 并发编程笔记
-
最近在看《Programming in Go》, 其中关于并发编程写得很不错, 受益非浅, 其中有一些例子是需要多思考才能想明白的, 所以我打算记录下来, 强化一下思路
《Programming in Go》在
Chapter 7. Concurrent Programming
里面一共用3个例子来讲述并发编程的3个模式, 第一个是filter
, 筛选出后缀名和文件大小文件列表, 还算简单就不说, 然后第二个是升级版, 正则版filter
, 不同的是他是根据正则搜索出文件的文本并且列出来. 这个例子我起初看是有点蒙的, 这样写是没错, 但是为什么要这样写, 他的设计思路是什么, 和其他方法相比他有什么优势, 这些都不清楚, 于是决定好好分析一下. 实际上这个例子实现的功能并不复杂, 所以我的文章实际上是在讨论怎么产生出和作者相似的思路.如果不考虑用 goroutine 的话, 思路其实很简单:
1. 列出文件列表, 编译正则. 2. 遍历文件, 打开并遍历每行, 如果正则能匹配, 记录下来. 3. 列出来.
如果用 goroutine , 就会有以下思路:
1. 在得到文件路径数组之后, 分发任务给N个核. 2. 每个核负责打开文件, 将符合条件的那行文本写入到 `channel` 3. 主线程等待并接收`channel`的结果. 显示出来, 完毕
** 然后下文才是重点 **
- channel关闭的时机
在go中, channel 是不会自动关闭的, 所以需要在我们使用完之后手动去关闭, 而且如果使用for语法来遍历channel每次得到的数据, 如果channel没有关闭的话会陷入死循环. 在 goruntine 中会造成 deadlock
for job := range jobs { fmt.Println(job) }
如果没close, 会触发dead lock. 因为for...range...会自动阻塞直到读取到数据或者channel关闭, 没close的话就会导致整个channel处于睡眠状态. channel关闭后, 就不允许写入(缓冲的数据还在, 还可以读取), 所以, channel 关闭的时机很重要.
- 分发任务
我所知道任务分发方法有两种:
第一种是固定分配, 如果说我想计算1+2+3+...+100, 然后分成4份, 也就是
1+2+..+25
,...
,...
,86+87+...+100
, 然后再将结果累加起来.还有一种是抢占式的, 这里需要使用一个队列, 将所有任务写入队列, 然后开N个goroutine, 每个goroutine从队列读取任务(要确保线程安全), 处理, 完成后再继续读取任务. 不再是固定分配, 自己那份做完了就休息了, 所以看来第二种要好一点.
采用第二种方式的话, 对应go的做法, 那就是使用一个channel, 命名为
jobs
, 将所有的任务写入进去, 写入完毕之后关闭这个 channel, 当然, 因为是N核, 系统能同时处理的任务我们设置为N个(也就是我们使用了N个goruntine), 那么声明jobs
是缓冲区长度为N的 channel.Buffered channel
和普通的 channel 的差别是他可以同时容纳多个单位数据, 当缓存的数据单位数量等于 channel 容量的时候, 再执行写入将会阻塞, 否则都是及时处理的.- 结果集
当我们将数据处理后, 就需要将结果收集起来. 需要注意的是, 这些操作不是在主 goruntine 执行, 所以我们需要通过 channel 传递给主 goruntine . 所以只需要在外部声明一个名为
results
的 channel . 然后在主 goruntine 通过for
来显示, 这时候就会发现一个问题, 这个results
关闭的时机问题. 正确的关闭时机是写入所有的Result
之后. 但是别忘了我们同时开了多个 goruntine , 所以results
应该在执行任务的 goruntine 完成信号累计到N个
这个时机关闭. 所以我们再引入一个名叫done
的 channel 来解决. 每个 goruntine 发送完 result 后会写入一次done, 然后我们就可以遍历 done , 遍历之后说明全部完成了, 再执行显示.Result 的数据结构
type Result struct { filename string lino int line string }
书中的
cgrep1
就是这样的func awaitCompletion(done <-chan struct{}, results chan Result) { for i := 0; i < workers; i++ { <-done } close(results) }
但是这样有可能造成死锁, 因为书中
results
缓冲区长度限定为最大1000个, 也就是超过1000个 result 的时候再打算写入 result 会等待取出 result 后才执行, done 也不会写入, 而awaitCompletion
是等到所有 goruntine 都完成了才会取出results
, 而且当result
非常大的时候因为内存的缘故也是不可能一次性取出的. 所以就需要在读取results
的同时读取done
, 当读取done
次数大于 N 后关闭results
, 所以, 因为要在多个 channel 中同时读取, 所以需要使用select
.下面是书中的
cgrep3
, 改进版:func waitAndProcessResults(timeout int64, done <-chan struct{}, results <-chan Result) { finish := time.After(time.Duration(timeout)) for working := workers; working > 0; { select { // Blocking case result := <-results: fmt.Printf("%s:%d:%s\n", result.filename, result.lino, result.line) case <-finish: fmt.Println("timed out") return // Time's up so finish with what results there were case <-done: working-- } } for { select { // Nonblocking case result := <-results: fmt.Printf("%s:%d:%s\n", result.filename, result.lino, result.line) case <-finish: fmt.Println("timed out") return // Time's up so finish with what results there were default: return } } }
看到这里, 我就有个疑问, 为什么在全部完成之后(done都接收到N个了), 还要再遍历出
results
, 直到读取不到才算读取完成呢(我反应一向比较慢^_^)? 于是我做了个实验, 去掉了后面再次循环的部分, 发现有时会遗漏掉数据(我用4个测试文件...), 证明这段代码是有用的!!!我的想法是, 他是在处理完 result, 然后写入
results
, 写完了才发送done
, 也就是在收到所有的 done 之后, 所有的数据应该是已经处理完成的. 为了验证这个想法, 我写了一下代码:for working := workers; working > 0; { select { // Blocking case result := <-results: // received result case <-done: working-- if working <= 0 { println(len(results)) } } }
然后看到输出的数是大于0的, 也就是说在接收到全部 done 之后,
results
还有数据在缓冲区中, 然后在看看发送result
的代码, 突然就明白了func doJobs(done chan<- struct{}, lineRx *regexp.Regexp, jobs <-chan Job) { for job := range jobs { job.Do(lineRx) } done <- struct{}{} }
我把写入和读取想当然认为一起发生了, 因为有缓冲区的缘故, doJobs在发送进
results
的缓冲区之后就立刻发送done
了, 但是写入的数据有没有被处理, 是不知道的, 所以在接收到所有done
之后,results
缓冲区还有数据, 需要再循环一遍.
附我的代码一份:
package main import ( "fmt" "os" "runtime" "regexp" "log" "bufio" ) type Job struct { filename string results chan<- Result } type Result struct { filename string line string lino int } var worker = runtime.NumCPU() func main() { // config cpu number runtime.GOMAXPROCS(worker) files := os.Args[2:] regex, err := regexp.Compile(os.Args[1]) if err != nil { log.Fatal(err) return } // 任务列表, 并发数目为CPU个数 jobs := make(chan Job, worker) // 结果 results := make(chan Result, minimum(1000, len(files))) // 标记完成 dones := make(chan struct{}, worker) go addJob(files, jobs, results) for i:=0; i<worker; i++ { go doJob(jobs, regex, dones) } awaitForCloseResult(dones, results) } func addJob(files []string, jobs chan<- Job, results chan<- Result) { for _, filename := range files { jobs <- Job {filename, results} } close(jobs) } func doJob(jobs <-chan Job, regex *regexp.Regexp, dones chan<- struct{}) { for job := range jobs { job.Do(regex) } dones <- struct{}{} } func awaitForCloseResult(dones <-chan struct{}, results chan Result) { working := worker done := false for { select { case result := <-results: println(result) case <-dones: working -= 1 if working <= 0 { done = true } default: if done { return } } } } func (j *Job) Do(re *regexp.Regexp) { f, err := os.Open(j.filename) if err != nil { println(err) return } defer f.Close() b := bufio.NewReader(f) lino := 0 for { line, _, err := b.ReadLine() if re.Match(line) { j.results <- Result {j.filename, string(line), lino} } if err != nil { break } lino += 1 } } func minimum(a, b int) int { if a > b { return b } return a } func println(o ...interface{}) { fmt.Println(o...) }