Go并发编程

目录

一些基本概念

并发任务单元的状态

并发任务单元：进程，线程，协程

同步

异步

并发和并行

并发编程

创建并发任务

WaitGroup

等待goroutine结束

WaitGroup.Wait

WaitGroup.Add

获取CPU数量

获取Goroutine的编号和返回值

GOMAXPROCS

重新调度

终止任务

终止进程：os.Exit

终止当前任务：runtime.Goexit

通道 Channel

声明

发送、接收数据

等待goroutine结束

同步模式和异步模式

指针

cap & len：获取缓冲区大小和当前已缓冲数量

关闭通道：close

关闭同步通道：解除阻塞

关闭异步通道

ok-idom模式

range模式

---

Go的并发绝对称得上是Go的一大特色。Go使用类似协程的方式来处理并发单元，却又在运行时层面做了更深度的优化处理。这使得语法上的并发编程变得极为容易，仅仅使用关键字go就可以创建一个goroutine（并发任务单元）。然而简便的并发带来的是控制上的难度，而Go的并发编程也足以写出一部篇幅不少的大作。本篇博客用较为浅显的例子总结下Go并发编程中常见的问题和解决方案。

一些基本概念

在开启Go并发编程之前，我们先来总结一些常见的概念。这些概念在并发编程中或多或少都会用到。

并发任务单元的状态

一个并发任务单元的生命周期，从新建开始，包括就绪、运行、阻塞和终止。阻塞指的是数据未准备就绪，该并发任务单元一直等待。例如某个运行中的并发任务单元需要使用一个资源，但不巧的是该资源被其它并发任务单元占用，因此该并发任务单元就会进入阻塞状态，一直等待该资源被释放。

并发任务单元：进程，线程，协程

进程、线程和协程是我们经常听到的三个概念。它们都是并发任务单元。

• 进程：进程是指一个程序在给定数据集合上的一次执行过程，是系统进行资源分配和运行调用的独立单位。可以简单的理解为操作系统中正在执行的程序。
• 线程：线程是由进程创建的。进程启动时会最先创建一个线程，即主线程。主线程可以创建其它的子线程，因此一个进程可以包含一个或多个线程。线程必须在某个进程中执行，一个进程内的多个线程共享该进程所拥有的所有数据资源，例如打开的文件、同一个地址空间、甚至是进程所拥有的硬件设备（物理内存、磁盘、打印机等）。
• 协程：协程也被称作微线程，它的资源开销比线程更小。而go关键字创建的并发任务单元可以简单的理解为是一个协程。

同步

在发起一个调用时，在没有得到结果之前，该过程会一直等待，直到该调用返回。例如调用一个函数，在该函数没有返回结果之前（哪怕该函数本身没有返回值），调用者会一直等待该函数返回（即执行结束）。

异步

调用者在发起一个调用后，不必等待该调用是否执行完毕，这就是异步。

并发和并行

我们经常提到并发和并行的概念，但经常容易混淆二者的意思。先来看概念：

并发：逻辑上具备同时处理多个任务的能力

并行：物理上在同一时刻执行多个并发任务

我们通常说的程序是并发设计的，指的是所设计的程序允许多个任务同时执行。但实际上往往并不是我们所期望的那样。例如在单核处理器上，多个任务只能以间隔的方式切换执行。并行依赖多核处理器等物理设备，也就是说，并行是并发设计的理想执行模式。

并发编程

在说完上面常见的名词之后，我们来看看Go的并发编程。

创建并发任务

只需在函数调用前添加关键字go即可实现并发任务单元goroutine的创建：

package main

import "fmt"

func main() {
	go fmt.Println("hello, world!")
	
	go func(message string) {
		fmt.Println(message)
	} ("hello world!")
}

需要注意的是关键字go并非执行并发操作，而是创建了一个并发任务单元。创建后任务会被放置在系统队列中，等待调度器安排合适的系统线程去获取执行权。并发任务单元在运行时不保证彼此之间的执行顺序。

当有多个逻辑处理器时，调度器会将 goroutine 平等分配到每个逻辑处理器上。 这会让 goroutine 在不同的线程上运行。 不过要想真的实现并行的效果，用户需要让自己的程序运行在有多个物理处理器的机器上。 否则，哪怕 Go语言运行时使用多个线程，goroutine 依然会在同一个物理处理器上并发运行，达不到并行的效果。

与defer一样，goroutine在创建时会立即计算并复制执行参数：

package main

import (
	"fmt"
	"time"
)

// 默认值是0。
var c int

func counter() int {
	c++
	return c
}

func main() {
	a := 100

	// 使用真实的值传入
	go func(x, y int) {
		// 利用time.Sleep将goroutine阻塞1秒，使goroutine的逻辑在main之后运行。
		// 这里不是一个好的方式来阻塞goroutine。后面章节会介绍更好的方案来控制goroutine的执行顺序。
		time.Sleep(time.Second)
		fmt.Println("goroutine1: ", x, y)
	}(a, counter())

	// 换成指针
	go func(x, y *int) {
		// 让该goroutine最后执行。
		time.Sleep(time.Second * 2)
		fmt.Println("goroutine2: ", *x, *y)
	}(&a, &c)

	a += 100
	fmt.Println("main: ", a, counter())

	c = 23

	// 等待两个goroutine执行结束。
	// 这里也不是一个推荐方法，后续章节会详细介绍如何等待goroutine结束。
	time.Sleep(time.Second * 3)

	// 程序输出
	// main:  200 2
	// goroutine1:  100 1
	// goroutine2:  200 23
}

WaitGroup

WaitGroup的常见作用是等待goroutine的结束。因为进程退出时不会等待goroutine结束，因此当main函数退出时，goroutine可能还没有开始执行：

package main

import (
	"fmt"
	"time"
)

func main() {
	go func() {
		time.Sleep(time.Second)
		fmt.Println("Inside the goroutine.")
	}()

	fmt.Println("exit...")
	
	// 输出：
	// exit...
}

上一个小节“创建并发任务”中，main函数使用time.Sleep来等待所有goroutine结束。这虽然算是一个行之有效的方式，但并不推荐。因为在实际开发中，我们并不会严格知道goroutine的所用时间。

等待goroutine结束

如要等待多个任务结束，sync.WaitGroup是一个推荐的选择。通过设定计数器，让每个goroutine在退出前递减，直至归零时解除阻塞。

package main

import (
	"fmt"
	"sync"
	"time"
)

var (
	max       = 10
	waitGroup sync.WaitGroup
)

func main() {
	// 累加计数
	waitGroup.Add(max)
	for i := 0; i < max; i++ {
		go func(index int) {
			// 递减计数
			defer waitGroup.Done()
            // 这里的阻塞是为了保证main函数中 “Inside main function”的打印先于所有goroutine执行
			time.Sleep(time.Second)
			fmt.Println("goroutine: ", index)
		}(i)
	}

	fmt.Println("Inside main function.")

	// 此时main阻塞，直到计数归零
	waitGroup.Wait()

	fmt.Println("main exit.")

	// 程序输出：
	// Inside main function.
	// goroutine:  7
	// goroutine:  1
	// goroutine:  3
	// goroutine:  8
	// goroutine:  0
	// goroutine:  2
	// goroutine:  6
	// goroutine:  5
	// goroutine:  9
	// goroutine:  4
	// main exit.
}

WaitGroup.Wait

Wait可以在多处阻塞，它们都能接收到通知。

在上面的例子中，我们为了保证goroutine在main函数"Inside main function"打印之后执行，在goroutine内部添加了time.Sleep。其实我们可以利用Wait的机制避免使用time.Sleep这种粗暴的手段：

package main

import (
	"fmt"
	"sync"
)

var (
	max       = 10
	waitGroup sync.WaitGroup
	start     sync.WaitGroup
)

func main() {
	// 累加计数
	start.Add(1)
	waitGroup.Add(max)
	for i := 0; i < max; i++ {
		go func(index int) {
			// 递减计数
			defer waitGroup.Done()
			// 保证main函数中 “Inside main function”的打印先于所有goroutine执行
			start.Wait()
			fmt.Println("goroutine: ", index)
		}(i)
	}

	fmt.Println("Inside main function.")
	start.Done()

	// 此时main阻塞，直到计数归零
	waitGroup.Wait()

	fmt.Println("main exit.")
}

WaitGroup.Add

尽管WaitGroup.Add实现了原子操作，但建议在goroutine外使用。以免Add尚未执行，Wait已经退出。

package main

import (
	"fmt"
	"sync"
)

var (
	waitGroup sync.WaitGroup
)

func main() {
	go func() {
		// 来不及设置
		waitGroup.Add(1)
		fmt.Println("Inside the goroutine.")
		waitGroup.Done()
	}()

	waitGroup.Wait()
	fmt.Println("main exit.")
	// 程序输出：
	// main exit.
}

获取CPU数量

runtime.NumCPU函数返回一个int型数据，表示当前执行机器的CPU数量。

package main

import (
	"fmt"
	"runtime"
)

func main() {
	fmt.Println(runtime.NumCPU())
}

获取Goroutine的编号和返回值

使用go关键字创建的goroutine无法像普通函数调用那样获取返回值。所创建的goroutine也不能获知并发任务的编号。这些问题我们可以使用本地存储的方式解决。

package main

import (
	"fmt"
	"sync"
)

var (
	size      = 10
	waitGroup sync.WaitGroup
)

// 存储goroutine ID和返回结果的本地存储
type LocalStorage struct {
	ID     int
	Result interface{}
}

func main() {
	pool := make([]LocalStorage, size)

	for i := 0; i < size; i++ {
		waitGroup.Add(1)

		go func(id int) {
			defer waitGroup.Done()

			// 使用id*3拟定一个执行结果
			pool[id].ID = id
			pool[id].Result = id * 3
		}(i)
	}

	waitGroup.Wait()

	fmt.Printf("%+v\n", pool)
	// [{ID:0 Result:0} {ID:1 Result:3} {ID:2 Result:6} {ID:3 Result:9} {ID:4 Result:12} {ID:5 Result:15} {ID:6 Result:18} {ID:7 Result:21} {ID:8 Result:24} {ID:9 Result:27}]
}