并发任务的固定模式,该如何处理
虽然我们可以自己使用 chan 来实现很多并发的框架和模式，但实际应用的时候先查查 sync 包总是没错的，比如 sync/atomic 包对原子操作进行了支持，上一个章节的 *x = *x +1，其实可以直接使用 atomic.AddUint64(&x, 1) 来实现。
标准库 sync.Once 可以实现只执行一次的功能，比如你指向创建一个对象(单件)。下面这个 born 函数永远只会造同一个人:
type person struct{}
var instance *person
var once sync.Once
func born() *person {
once.Do(func() {
fmt.Println("构建 instance 对象")
instance = &person{}
})
return instance
}
我们启动 10 个 goroutine 来测试一下:
func getInstance() {
wg := sync.WaitGroup{}
wg.Add(10)
for i := 0; i < 10; i++ {
go func() {
defer wg.Done()
instance := born()
fmt.Printf("%p\n", instance) // 0x1254790
}()
}
wg.Wait()
}
每次获取到的 instance 的地址都是一样的，说明是同一个对象，它的实现机制如下:
func (o *Once) Do(f func()) {
if atomic.LoadUint32(&o.done) == 0 {
o.doSlow(f)
}
}
func (o *Once) doSlow(f func()) {
o.m.Lock()
defer o.m.Unlock()
if o.done == 0 {
defer atomic.StoreUint32(&o.done, 1)
f()
}
}
首先是对通过 mutex 和 atomic.StoreUint32 配合来实现，看到这里 doSlow 函数就奇怪了，你不是有 mutex 吗? 为什么还要来一发 aotmic 的函数操作呢? doSlow 函数名其实已经说明了一切。doSlow 说明这个函数很慢，因为 mutex 是一个成本很高的操作，而整数的原子读写却轻量很多，所以 atomic.LoadUint32 先判断了一下 o.done 是不是为 0，不是的话就滚蛋了，少量的竞争者最后能到达 if o.done == 0 的判断，这个提前的判断提高了性能(请注意 defer 语句执行的顺序，对 mutex 解锁 Unlock 比设置 o.done 为 1 的操作后执行。
sync/atomic 不仅能对基本的数值类型进行原子操作，atomic.Value 还能对 interface{} 数据类型进行操作。
既然系统包这么好用，为什么不把所有的应用模式封装完呢? 有时候系统包并不好用，看下面的 mutex 的例子:
func main() {
var mu sync.Mutex
go func(){
fmt.Println("NO.1中国人")
mu.Lock()
}()
mu.Unlock()
}
程序的意图简洁清晰明了，看起来好像没有什么问题，但是执行大概率是报错:
fatal error: sync: unlock of unlocked mutex
因为 go func 的 goroutine 和 main 的 goroutine 执行顺序并不能保证谁先谁后，mu.Unlock 的时候 mu.Lock 还未执行，所以 Unlock 直接抛出了错误。要修改它也很麻烦，要这样写:
func main() {
var mu sync.Mutex
mu.Lock
go func(){
fmt.Println("NO.1中国人")
mu.Unock()
}()
mu.lock()
}
是不是不好理解? 第二个 mu.Lock 的时候会等着 mu.Unlock 先执行，有点绕口。但是如果使用无缓冲的通道来重构代码不但简单而且很可靠:
func main() {
exit := make(chan int)
go func() {
fmt.Println("NO.1中国人")
<-exit
}()
exit <- 1
}
为什么很可靠? 因为 exit 没有缓冲，不管 exit <- 1 先执行到 还是 <-exit 先执行到，都得等待同步交接，咱们谁先到都见面聊。这个例子中 <- exit 和 exit <- 1 的位置可以交换，他们的读写本身并没有意义，chan 的类型也是无意义的，用 string、struct{} 都行，但是 chan 完成了比 mutex 很易懂的代码。把这个例子扩展一下，启动指定数量的任务:
func lockBatch(taskCount int) {
waitGroup := make(chan struct{}, taskCount)
for i := 0; i < cap(waitGroup); i++ {
go func(index int) {
fmt.Printf("NO.1中国人%d\n", index)
waitGroup <- struct{}{}
}(i)
}
for i := 0; i < cap(waitGroup); i++ {
<-waitGroup
}
}
这不就是 sync.WaitGroup 干的事情吗? 这就是框架，或者说模式。
这个模式可以启动 N 个协程，来做 N 件事情，每个协程和任务的比例是 1:1，不过太多的协程可能不利于管理和回收，能不能抽象出 N 个协程做 M 件事情的模式呢? 为了让 goroutine 可以不断的做任务，可以使用一个有缓冲的通道来保存任务，极端的情况是只有一个 goroutine 来做任务，它不断的从缓冲通道里取出任务执行，知道通道关闭为止，由于不同的 goroutine 可以安全的从通道中读取任务，因此可以随意调节 goroutine 的数目而不用修改代码。定一个 handle 函数表示做任务的工人:
var wg sync.WaitGroup
func handle(tasks chan string, index int) {
defer wg.Done()
for {
// 读取关闭的无缓冲通道，返回零值和false
// 读取关闭的有缓冲通道，先把缓存数据读完后，再读取返回零值和false
task, ok := <-tasks
if !ok {
// 任务通道关闭了，不会再有新任务了
fmt.Printf("handle %d over\n", index)
return
}
fmt.Printf("干活儿 %s By 工人 %d\n", task, index)
}
}
tasks 表示任务的需求，如果只需要一个工人，直接 go handle 就行了，多个工人只需一个循环，工人下班的标志依靠 wg 来保证，干完活就能下班，而 tasks 通道用来接收任务。
func createTasks(goroutines, taskCount int) {
tasks := make(chan string, taskCount)
// 启动 N
wg.Add(goroutines)
for i := 0; i < goroutines; i++ {
go handle(tasks, i)
}
// 设置 M
for k := 0; k <= taskCount; k++ {
tasks <- fmt.Sprintf("任务 %d", k)
}
close(tasks)
wg.Wait()
}
参数 goroutines 是工人的个数(协程的个数)，而 taskCount 是任务的数量。需要注意的时候，在通道 close 关闭后，通道中的数据依然可以读取直到读完后再次读取事 task 是零值(这里是空字符串，因为 chan 的类型是 string)，ok 为 false，也就是没有任务了，工人都可以下班了(wg.Done)，而 wg.Wait 会等所有工人都下班后，才把工厂的门关上(函数返回)。如此，就完成了 N:M 的任务模型，这里 N 和 M 可以通过调用参数传递数量，而你不用修改内部的代码。
进一步延伸这个例子，就是生产者和消费者，而生产者和消费者的数量可以是任何比例。
func producer(name string, tasks chan<- string) {
index := 0
for {
tasks <- fmt.Sprintf("%s-task-%d", name, index)
time.Sleep(50 * time.Millisecond)
index++
}
}
func consumer(name string, tasks <-chan string) {
for task := range tasks {
fmt.Printf("%s-%s has been token\n", name, task)
time.Sleep(50 * time.Millisecond)
}
}
producer 生产者是个是循环，源源不断的造东西，而 consumer 一直从 tasks 里取货，模拟一个工厂的代码:
func createFactory(stockSize, producerCount, consumerCount int) {
tasks := make(chan string, stockSize)
for i := 0; i <= producerCount; i++ {
go producer(fmt.Sprintf("p%d", i), tasks)
}
for i := 0; i <= consumerCount; i++ {
go consumer(fmt.Sprintf("c%d", i), tasks)
}
exit := make(chan os.Signal, 1)
// ctrl+c 中断程序
signal.Notify(exit, syscall.SIGINT, syscall.SIGTERM)
<-exit
}
stockSize 是车间大小，表示了最多能存多少货，producerCount 是生产者个数，consumerCount 是消费者个数，当生产者 pdoducer 足够多的时候，把仓库 stockSize 塞满了就阻塞了没法再生产了，除非消费者 consumer 来拿走一些；反之如果消费者足够多的时候，仓库里没货也只能等着了，除非生产者 producer 造点货出来。
为了达到平衡，如果希望生产者以稳定的效率工作，不多不少，如何控制呢? 非常简单:
func producer2(max int, tasks chan<- string, doSth func() string) {
lines := make(chan int, max)
for {
lines <- 1
tasks <- doSth()
<-lines
}
}
max 表示最大的生产线，每次开一个生产线，就往通道里计数一个，生产完了，再从通道取出一个，因为通道最大是 max，满了就阻塞了，随时都保证了只有 max 条生产线存在。
下面的这个例子实现的功能是，等待一个对象完成自己的工作，和前面不同的是，它把 chan 的状态作为对象的一部分:
type request struct {
data []interface{} // 模拟要处理的数据
finish chan struct{} // 完成后的标志
}
func newRequest(data ...interface{}) *request {
return &request{data, make(chan struct{}, 1)}
}
func doWork(req *request) {
time.Sleep(1 * time.Second) // 模拟耗时
fmt.Println(req.data) // 工作内容就是打印
close(req.finish) // req.finish <- struct{}{} 也行
}
func createWork() {
req := newRequest(1, "string", true)
go doWork(req)
for i := 0; i < 100; i++ {
time.Sleep(50 * time.Millisecond)
fmt.Println("继续做事，我很忙")
}
<-req.finish
}
createWork 函数通过并发提高了处理效率，单独开了一条线去处理 request，同时它可以继续往后做事，但是它保证在函数返回前 req 对象肯定被处理了。关于 chan 和并发，还有很多有用的固定模式，希望这几个例子可以抛砖引玉。
本章节的代码 https://github.com/developdeveloper/go-demo/tree/master/16-concurrent-pattern
~小贴士：记得定期检查您的路由器设置并更新您的Wi-Fi密码，以防止网络被攻击。同时，考虑在您的网络中添加一些额外的安全措施，例如启用双因素认证。
