什么是协程?
进程和线程
一个应用程序时运行在操作系统上的一个进程。进程是一个运行在自己独立内存空间的独立执行体,是操作系统进行资源分配的最小单位。一个进程则有一个或多个线程组成,这些线程是共享进程内存地址空间的执行体,是操作系统进行任务调度的最小单位。而使用多线程进行工作时,由于共享父进程的内存空间等资源,访问同一个数据需要对其进行加锁,保证同一时间只有一个线程操作一个数据。这样不仅会提高编码的复杂度,还会有多个线程抢占锁、线程切换带来的额外开销。
协程
在Go中,应用程序并发处理的部分被称作goroutines(协程),它是一种更轻量级的线程,和操作系统的线程之间并无一对一的关系。协程是根据一个或多个线程的可用性,映射(多路复用,执行于)在它们之上的;协程调度器负责在Go运行时调度进行协程的工作。
通道(Channel)
协程工作在相同的地址空间中,所以共享内存的方式是同步的,可以使用互斥锁来实现,但是Go中更好的方案是使用Channel来同步协程。
通道类型(Chan)就像一个可用于发送类型化数据的管道,由其负责协程之间的通信,在任何时间,一个通道数据被设计为只有一个协程可以对其访问,所以不会发生数据竞争。
通道阻塞
默认情况下,Go创建的通道是同步且无缓冲的:在有接受者接受数据之前,发送不会结束,发送者是直接将数据交给接受者的,所以这种通道的发送或接受操作在对方准备好之前都是阻塞的。
例如以下代码,执行会报错死锁:
示例1.1:
func main() {
ch := make(chan int)
ch <- 1
<-ch
}
因为对ch的读写都在main函数的主协程中,执行到ch <-1
时由于接收ch的数据还没准备好,发送数据将被阻塞,程序无法继续执行。必须使用关键字go
来启动一个新的协程发送数据,另一个协程接收数据,如下所示:
示例1.2
func main() {
ch := make(chan int)
go func() {
ch <- 1
}()
fmt.Println(<- ch)
}
使用make
创建一个通道的时候可以传入第二个参数指定通道缓冲区大小,这种方式在通道写满之前,发送数据不会被阻塞,通道不为空时接收操作不会被阻塞,如果将示例1.1中的创建通道传第二个参数为1,就可以正常运行不会死锁了,代码如下:
示例1.3
func main() {
ch := make(chan int, 1)
ch <- 1
fmt.Println(<- ch)
}
Go协程调度原理
调度器架构
Go的调度器从最开始的单线程经过不断的改进、优化,发展到现在的GMP模型,在GMP模型中有三个重要的结构:
- G(Goroutine):go协程,一个可执行单元,调度器作用就是对所有G的切换
- M(Thread):操作系统上的线程,G运行与M上,一个G可能由多个不同的M运行,一个M可以运行多个G
- P(Processor):处理器,他包含了运行G的资源,如果线程M想运行G,必须先获取P,P还包含了可运行的G队列。一个M一个时刻只拥有一个P,M和P的数量是1:1的。
上图中各个模块的作用如下:
- 全局队列:存放等待运行G
- P的本地队列:和全局队列类似,存放的也是等待运行的G,存放数量上限256个。新建G时,G优先加入到P的本地队列,如果队列满了,则会把本地队列中的一半G移动到全局队列
- P列表:所有的P都在程序启动时创建,保存在数组中,最多有
GOMAXPROCS
个,可通过runtime.GOMAXPROCS(N)
修改,N表示设置的个数
M是Goroutine调度器和操作系统调度器的桥梁,每个M代表一个内核线程,操作系统调度器负责把内核线程分配到CPU的核心上执行。
调度策略
复用线程
调度器核心思想是尽可能避免频繁的创建、销毁线程,对线程进行复用以提高效率。
1. work stealing机制(窃取式)
当本线程无G可运行时,尝试从其他线程绑定的P窃取G,而不是直接销毁线程。
2. hand off机制
当本线程M因为G进行的系统调用阻塞是,线程释放绑定的P,把P转移给其他空闲的M'执行。
利用多核CPU并行
GOMAXPROCS
设置P的数量,最多有GOMAXPROCS
个线程分布在多个CPU核心上运行。
抢占
一个goroutine最多占用CPU10ms,防止其他goroutine等待太久得不到执行被“饿死”。
全局G队列
全局G队列是有互斥锁保护的,访问需要竞争锁,新的调度器将其功能弱化了,当M执行work stealing从其他P窃取不到G时,才会去全局G队列获取G。
Go并发编程实例
两个协程交替打印1-100
用两个协程顺序打印出1-100,要求一个协程打印1、3、5、7...奇数,另一个协程打印2、4、6、8...偶数,输出必须是顺序的。
示例代码:
func main() {
// ch用来同步两个协程交替执行
ch := make(chan int)
// ch_end用来阻塞主程序,让两个协程可以执行完
ch_end := make(chan int)
go func() {
for i := 1; i <= 100; i++ {
ch <- 1
if i % 2 == 0 {
fmt.Println(i)
}
}
ch_end <- 1
}()
go func() {
for i := 1; i <= 100; i++ {
<-ch
if i % 2 != 0 {
fmt.Println(i)
}
}
}()
<-ch_end
}
并行素数筛选
有一个协程不断生2~n的自然数,对每个素数起一个协程,用来筛选素数
示例代码:
func generate(ch chan int, n int) {
for i := 2; i <= n ; i++ {
fmt.Println("generate:", i)
ch <- i
}
close(ch)
}
func filter(in, out chan int, prime int) {
for i := range in {
fmt.Printf("filter(%d): %d\n", prime, i)
if i % prime != 0 {
out <- i
}
}
close(out)
}
func main() {
res := []int{}
ch := make(chan int)
go generate(ch, 112)
for {
if prime, ok := <- ch; ok {
res = append(res, prime)
ch_out := make(chan int)
go filter(ch, ch_out, prime)
// 前一个素数过滤协程的输出通道是后一个素数过滤通道的输入通道
ch = ch_out
} else {
break
}
}
fmt.Println("main:", res)
}
实现超时机制
当设置的超时时间到达后如果work还不可执行就终止等待,返回超时
示例代码
func TimeOut(timeout time.Duration) {
ch_to := make(chan bool, 1)
go func() {
time.Sleep(timeout)
ch_to <- true
}()
ch_do := make(chan int, 1)
go func() {
time.Sleep(3e9)
ch_do <- 1
}()
select {
case i := <- ch_do:
fmt.Println("do something, id:", i)
case <-ch_to:
fmt.Println("timeout")
break
}
}
实现迭代器
实现一个惰性迭代器,每次调用返回一个列表元素,直到所有的元素被访问完返回nil
示例代码:
// 迭代器
func iterator(iterable []interface{}) chan interface{}{
yield := make(chan interface{})
go func() {
for i := 0; i < len(iterable); i++ {
yield <- iterable[i]
}
close(yield)
}()
return yield
}
// 获取下一个元素
func next(iter chan interface{}) interface{} {
for v := range iter {
return v
}
return nil
}
func main() {
nums := []interface{}{1, 2, 3, 4, 5}
iter := iterator(nums)
for v := next(iter); v != nil; v = next(iter) {
fmt.Println(v)
}
}