Go实践之并发初体验

²⁸/Nov 2014

golang的一大卖点是并发模型基于CPS(continuation passing style)，使用起来比较简单。刚开始我也觉得比较简单，但使用之后发现，你必须很清楚golang的并发机制才能使用自如，在需要同步尤甚，一不小心就会陷入罪恶的渊薮。

如何我们想从1顺序打印到10000，可能会这样写：

package main

import (
	"fmt"
)

var messages chan int
var done chan bool

func main() {
	messages = make(chan int)
	done = make(chan bool)
	times := 10000
	for i := 0; i < times; i++ {
		go func() {
			messages <- i
			done <- true
		}()
	}

	go func() {
		for i := range messages {
			fmt.Println(i)
		}
	}()

	for i := 0; i < times; i++ {
		<-done
	}
}

但事实上，打印的结果的是

10000
10000
....

由于10000个for循环执行地相当快，很可能在for已经循环结束， i的值增长到10000时， gorountines的调度才完成，i此时作为逃逸变量为goroutines共享，所以gorountines看到的i都是10000,打印的也都是10000。

以上共享变量的方式我们称之为Share，要解决这个问题，只能把main的i通过消息传递的方式Communicate给每个gorountines 例子：

	for i := 0; i < times; i++ {
		go func(v int) {
			messages <- v
			done <- true
		}(i)
	}

运行一下，结果是正确的。这时候才真正体会到那句话的奥妙：

Don’t communicate by shared memory. Instead, share memory by communicating. 

	                                                                             —— Rob Pike

Synchronization

golang没有join，无法直接在程序中设置等待点。实现同步通常有两种做法：

使用channel

由于channel分为阻塞和非阻塞的，使用阻塞的channel时就能达到同步的目的。上面两个例子都使用了channel进行同步。

使用WaitGroup

示例如下

func main() {
	var wg sync.WaitGroup
	wg.Add(1000)

	for i := 0; i < 1000; i++ {
		go func(v int) {
			defer wg.Done()
			fmt.Println(v)
		}(i)
	}
	wg.Wait()
	fmt.Println("exit")
}

Add操作设置要等待的gorountine个数，每执行一次Done，waitgroup就会减1, 执行Wait的地方就相当于join。

两种操作的本质都是设置一个计数器，每个gorountine执行完都会通知一下主程序，直到所以gorountine完全dead，main才会往下走。

当不清楚gorountines的数目时，在网上看到的一种做法是可以设置一个远大于gorountines的数目的阈值：

L: for { 
           select { 
               case <- done:
                   i++ 
                   if i > 10000 {
                         break L
                            }
                   }
            }

Concurrent Data Structure

在并发环境下，数据结构往往也需要设计成并发的，比如一个并发的Map可以这样设计：

type CorrMap struct {
	line2BusId  map[string][]string
	sync.RWMutex
}

func NewCorrMap() *CorrMap {
	return &CorrMap{line2BusId :  map[string][]string{}}
}

func (c *CorrMap) Add(line string, busId string) {
	c.Lock()
	defer c.Unlock()
	busIds, exists := c.line2BusId[line]
	if exists {
		c.line2BusId[line] = append(busIds, busId)
	} else {
		var tmp []string
		c.line2BusId[line] = append(tmp, busId)
	}
}

func (c CorrMap) Size() int {
	count := 0
	for _, v := range c.line2BusId {
		count+= len(v)
	}
	return count
}

不过利用Lock可能没有充分发挥golang的优势，更好的方法可以参考这篇博客。

Tags// concurrent, golang,