Go CSP并发模型

Posted 2023-04-23

参考技术A

Go的CSP并发模型

Go实现了两种并发形式。第一种是大家普遍认知的：多线程共享内存。其实就是Java或者C++等语言中的多线程开发。另外一种是Go语言特有的，也是Go语言推荐的：CSP（communicating sequential processes）并发模型。

CSP 是 Communicating Sequential Process 的简称，中文可以叫做通信顺序进程，是一种并发编程模型，由 Tony Hoare 于 1977 年提出。简单来说，CSP 模型由并发执行的实体（线程或者进程）所组成，实体之间通过发送消息进行通信，这里发送消息时使用的就是通道，或者叫 channel。CSP 模型的关键是关注 channel，而不关注发送消息的实体。 Go 语言实现了 CSP 部分理论 。

“ 不要以共享内存的方式来通信，相反， 要通过通信来共享内存。”

Go的CSP并发模型，是通过 goroutine和channel 来实现的。

goroutine 是Go语言中并发的执行单位。其实就是协程。

channel是Go语言中各个并发结构体(goroutine)之前的通信机制。 通俗的讲，就是各个goroutine之间通信的”管道“，有点类似于Linux中的管道。

Channel

Goroutine

Go 并发编程模型

一、前言

Go 语言中实现了两种并发模型，一种是依赖于共享内存实现的线程-锁并发模型，另一种则是CSP（Communicationing Sequential Processes，通信顺序进程）并发模型。

大多数编程语言（比如 C++、Java、Python 等）的并发逻辑都是基于操作系统的线程。并发执行单元（线程）之间的通信利用的就是操作系统提供的线程或进程间通信的原语，如共享内存、信号、管道、消息队列、套接字等。在这些通信原语中，使用最广泛的就是共享内存。 而 Go 语言从设计之初，就将解决上面传统并发模型问题作为目标，并在新并发模型设计中借鉴注明的 CSP并发模型。

• 线程与锁并发模型，类似于对底层硬件运行过程的形式化，程序的正确运行很大程度依赖于开发人员的能力和技巧，程序在出错时不易排查，写起代码来比较复杂，特别是在大型复杂的业务场景中，主要是 Go sync 包中的互斥锁、读写锁、条件变量、原子操作等同步原语。
• CSP模型，目的在于简化并发程序的编写，让并发程序的编写顺序像编写顺序程序一样简单，是 Go 的主流并发模型。

二、夯实基础

2.1 CSP 理论

CSP 并发模型最初由 Tony Hoare 于 1977 年的论文中被提出，它倡导 不要通过共享内存来通信，而要通过通信来实现内存共享。 这句话也是 Go 的并发哲学，它依赖 CSP 模型，基于 channel 实现。

CSP 模型中存在两个关键的概念：

• 并发实体，通常可以理解为执行线程，它们相互独立，且并发执行；
• 通道，并发实体之间使用通道发送信息。

与共享内存的线程与锁并发模型不同，CSP 中的并发实体是独立的，它们之间没有共享的内存空间。并发实体之间的数据交换通过通道实现，无论在通道中放数据还是从通道中取数据，都会导致并发实体的阻塞，直到通道中的数据被取出或者通道中被放入新的数据，并发实体通过这种方式实现同步。

CSP 类似于我们常用的同步队列，它关注的是消息传输的方式，即通道，消息的具体发送实体和具体接收实体并不关注。发送和接收信息的并发实体可能不知道对方具体是谁，它们之间是互相解耦的。通道与并发实体也不是紧耦合的，通道可以独立地进行创建和放取，并在不同的并发实体中传递使用。

CSP 通道的特性给并发编程提供了极大的灵活性，通道作为独立的对象，可以被任意创建、读取、放入数据，并在不同的并发实体中被使用。但是它也极易导致死锁，如果一个并发实体在读取一个永远没有数据放入的通道或者把数据放入一个永远不会被读取的通道中，那么它将被永远阻塞。

• 常见的线程模型：常见的有用户级线程模型、内核级线程模型、两级线程模型

2.2 线程模型

线程之间的调度永远是一个复杂的话题，但是并发编程必然会涉及到操作系统对线程的调度。根据资源访问权限的不同，操作系统会把内存分为内核空间和用户空间；

• 内核空间： 内核空间的指令代码具备直接调度计算机底层资源的能力，比如说 I/O 资源等；
• 用户空间： 用户空间的代码没有访问计算底层资源的能力，需要通过系统调用等方式切换为内核态来实现对计算机底层资源的申请和调度。

线程作为操作系统能够调度的最小单位，也分为用户线程和内核线程：

• 用户线程 由用户空间的代码创建、管理和销毁，线程的调度由用户空间的线程库完成（可能是编程语言层次的线程库），无需切换内核态，资源消耗少且高效。对 CPU 的竞争是以所属进程的维度参与的，同一进程下的所有用户级线程只能分时复用进程被分配的 CPU 时间片，所以无法很好利用 CPU 多核运算的优势。我们一般情况下说的线程其实是指用户线程；
• 内核线程 由操作系统管理和调度，能够直接操作计算机底层的资源，线程切换的时候 CPU 需要切换到内核态。它能够很好利用多核 CPU 并行计算的优势，开发人员可以通过系统调用使用内核线程。

用户线程是无法被操作系统感知的，用户线程所属的进程或者内核线程才能被操作系统直接调度，分配 CPU 的使用时间。对此衍生出了不同的线程模型，它们之间对 CPU 资源的使用程度各有千秋。

1. 用户级线程模型

用户级线程模型中基本是一个进程对应一个内核线程，如下图所示：

进程内的多线程管理由用户代码完成，这使得线程的创建、切换和同步等工作显得异常轻量级和高效，但是这些复杂的逻辑需要在用户代码中实现，一般依赖于编程语言层次。同时进程内的多线程无法很好利用 CPU 多核进程的优势，只能通过分时复用 的方式轮换执行。当进程内的任意进程阻塞，比如：线程 A 请求 I/O 操作被阻塞，很可能导致整个进程范围内的阻塞，因为此时进程对应内核线程因为线程 A 的I/O 阻塞而被剥夺 CPU 执行时间，导致整个进程失去了在 CPU 执行代码的权利！

2. 内核级线程模型

内核级线程模型中，进程中的每个线程都会对应一个内核线程，如下图所示：

进程内每创建一个新的线程都会调用操作系统的线程库在内核创建一个新的内核线程与对应，线程的管理和调度有操作系统负责，这将导致每次线程切换上下文时都会从用户态切换到内核态，会有不小的资源消耗，同时创建线程的数量也会受制于操作系统内核创建可创建的内核线程数量。好处是： 多线程能够充分利用 CPU 的多核并行计算能力，因为每个线程可以独立被操作系统调度分配到 CPU 上执行指令，同时某个线程的阻塞并不会影响到进程内其他线程工作的执行。

3. 两级线程模型

两级线程模型相当于用户级线程模式和内核级线程模型的结合，一个进程将会对应多个内核线程，由进程内的调度器决定进程内的线程如何与申请的内核线程对应，如下图所示：
进程会预先申请一定数量的内核线程，然后将自身创建的线程与内核进程进行对应。线程的调用和管理由进程内的调度器进行，而内核线程的调度和管理由操作系统负责。这种线程模型即能够有效降低线程创建和管理的资源消耗，也能够很好提供线程并行计算的能力。两级线程模型也给开发人员带来较大的技术挑战，因为开发人员需要在程序代码中模拟线程调度的细节，包括：线程切换时上下文信息的保存和恢复，栈空间大小的管理等。

三、应用实践

3.1 goroutine 的执行规则

《Go语言核心36讲》- go语句及其执行规则（极客时间）

在Go 语言中，不但有着独特的并发编程模型，以及用户级线程 goroutine，还拥有强大的用于调度 goroutine、对接系统级线程的调度器。

这个调度器是 Go 语言运行时系统的重要组成部分，它主要负责统筹调配 Go 并发编程模型中的三个主要元素，即：G（goroutine 的缩写）、P（processor 的缩写）和 M（machine 的缩写）。

其中的 M 指代的就是系统级线程。而 P 指的是一种可以承载若干个 G，且能够使这些 G 适时地与 M 进行对接，并得到真正运行的中介。

从宏观上说，G 和 M 由于 P 的存在可以呈现出多对多的关系。

• 当一个正在与某个 M 对接并运行着的 G 需要因某个事件（比如等待 I/O 或锁的解除）而暂停运行的时候，调度器总会及时地发现，并把这个 G 与那个 M 分离开，以释放计算资源供那些等待运行的 G 使用。
• 当一个 G 需要恢复运行的时候，调度器又会尽快地为它寻找空闲的计算资源（包括 M）并安排运行。另外，当 M 不够用时，调度器会帮我们向操作系统申请新的系统级线程，而当某个 M 已无用时，调度器又会负责把它及时地销毁掉。

正因为调度器帮助我们做了很多事，所以我们的 Go 程序才总是能高效地利用操作系统和计算机资源。程序中的所有 goroutine 也都会被充分地调度，其中的代码也都会被并发地运行，即使这样的 goroutine 有数以十万计，也仍然可以如此。

下面，我会从编程实践的角度出发，以go语句的用法为主线，向你介绍go语句的执行规则、最佳实践和使用禁忌。

M、P、G 之间的关系（简化版）

面试题1：什么是主 goroutine，它与我们启用的其他 goroutine 有什么不同？

我们具体来看一道我在面试中经常提问的编程题，下面打印结果是什么？

package main

import "fmt"

func main() 
  for i := 0; i < 10; i++ 
    go func() 
      fmt.Println(i)
    ()
  

这个程序很简单，三条语句逐条嵌套。我的具体问题是：这个命令源码文件被执行后会打印出什么内容？

这道题的典型回答是：不会有任何内容被打印出来。

问题解析

想必你已经知道，每条go语句一般都会携带一个函数调用，这个被调用的函数常常被称为go函数。而主 goroutine 的go函数就是那个作为程序入口的main函数。

一定要注意，go函数真正被执行的时间，总会与其所属的go语句被执行的时间不同。当程序执行到一条go语句的时候，Go 语言的运行时系统，会先试图从某个存放空闲的 G 的队列中获取一个 G（也就是 goroutine），它只有在找不到空闲 G 的情况下才会去创建一个新的 G。

这也是为什么我总会说“启用”一个 goroutine，而不说“创建”一个 goroutine 的原因。已存在的 goroutine 总是会被优先复用。

然而，创建 G 的成本也是非常低的。创建一个 G 并不会像新建一个进程或者一个系统级线程那样，必须通过操作系统的系统调用来完成，在 Go 语言的运行时系统内部就可以完全做到了，更何况一个 G 仅相当于为需要并发执行代码片段服务的上下文环境而已。

在拿到了一个空闲的 G 之后，Go 语言运行时系统会用这个 G 去包装当前的那个go函数（或者说该函数中的那些代码），然后再把这个 G 追加到某个存放可运行的 G 的队列中。

这类队列中的 G 总是会按照先入先出的顺序，很快地由运行时系统内部的调度器安排运行。虽然这会很快，但是由于上面所说的那些准备工作还是不可避免的，所以耗时还是存在的。

因此，go函数的执行时间总是会明显滞后于它所属的go语句的执行时间。当然了，这里所说的“明显滞后”是对于计算机的 CPU 时钟和 Go 程序来说的。我们在大多数时候都不会有明显的感觉。

在说明了原理之后，我们再来看这种原理下的表象。 请记住，只要go语句本身执行完毕，Go 程序完全不会等待go函数的执行，它会立刻去执行后边的语句。这就是所谓的异步并发地执行。

这里“后边的语句”指的一般是for语句中的下一个迭代。然而，当最后一个迭代运行的时候，这个“后边的语句”是不存在的。

案例中那条for语句会以很快的速度执行完毕。当它执行完毕时，那 10 个包装了go函数的 goroutine 往往还没有获得运行的机会。

请注意，go函数中的那个对fmt.Println函数的调用是以for语句中的变量i作为参数的。你可以想象一下，如果当for语句执行完毕的时候，这些go函数都还没有执行，那么它们引用的变量i的值将会是什么？

它们都会是10，对吗？那么这道题的答案会是“打印出 10 个10”，是这样吗？

在确定最终的答案之前，你还需要知道一个与主 goroutine 有关的重要特性，即：一旦主 goroutine 中的代码（也就是main函数中的那些代码）执行完毕，当前的 Go 程序就会结束运行。

如此一来，如果在 Go 程序结束的那一刻，还有 goroutine 未得到运行机会，那么它们就真的没有运行机会了，它们中的代码也就不会被执行了。

我们刚才谈论过，当for语句的最后一个迭代运行的时候，其中的那条go语句即是最后一条语句。所以，在执行完这条go语句之后，主 goroutine 中的代码也就执行完了，Go 程序会立即结束运行。那么，如果这样的话，还会有任何内容被打印出来吗？

严谨地讲，Go 语言并不会去保证这些 goroutine 会以怎样的顺序运行。由于主 goroutine 会与我们手动启用的其他 goroutine 一起接受调度，又因为调度器很可能会在 goroutine 中的代码只执行了一部分的时候暂停，以期所有的 goroutine 有更公平的运行机会。

所以哪个 goroutine 先执行完、哪个 goroutine 后执行完往往是不可预知的，除非我们使用了某种 Go 语言提供的方式进行了人为干预。然而，在这段代码中，我们并没有进行任何人为干预。

那答案到底是什么呢？就案例中如此简单的代码而言，绝大多数情况都会是“不会有任何内容被打印出来”。

但是为了严谨起见，无论应聘者的回答是“打印出 10 个10”还是“不会有任何内容被打印出来”，又或是“打印出乱序的0到9”，我都会紧接着去追问“为什么？”因为只有你知道了这背后的原理，你做出的回答才会被认为是正确的。

这个原理是如此的重要，以至于如果你不知道它，那么就几乎无法编写出正确的可并发执行的程序。如果你不知道此原理，那么即使你写的并发程序看起来可以正确地运行，那也肯定是运气好而已。

3.2 怎样才能让主 goroutine 等待其他 goroutine？

一旦主 goroutine 中的代码执行完毕，当前的 Go 程序就会结束运行，无论其他的 goroutine 是否已经在运行了。那么，怎样才能做到等其他的 goroutine 运行完毕之后，再让主 goroutine 结束运行呢？

其实有很多办法可以做到这一点，比如：

1. 主 goroutine Sleep “小睡”一会儿。
2. 先创建一个通道，它的长度应该与我们手动启用的 goroutine 的数量一致。在每个手动启用的 goroutine 即将运行完毕的时候，我们都要向该通道发送一个值。
3. sync.WaitGroup包控制

下面分别看一下优劣势：

主 goroutine Sleep “小睡”一会儿。

	for i := 0; i < 10; i++ 
		go func(i int) 
			fmt.Println(i)
		(i)
	
	time.Sleep(time.Millisecond * 500)

这个办法是可行的，只要“睡眠”的时间不要太短就好。不过，问题恰恰就在这里，我们让主 goroutine“睡眠”多长时间才是合适的呢？如果“睡眠”太短，则很可能不足以让其他的 goroutine 运行完毕，而若“睡眠”太长则纯属浪费时间，这个时间就太难把握了。你可能会想到，既然不容易预估时间，那我们就让其他的 goroutine 在运行完毕的时候告诉我们好了。这个思路很好，但怎么做呢？

使用通道发送信号

func main() 
	num := 10
	sign := make(chan struct, num)

	for i := 0; i < num; i++ 
		go func() 
			fmt.Println(i)
			sign <- struct
		()
	

	// 办法1。
	//time.Sleep(time.Millisecond * 500)

	// 办法2。
	for j := 0; j < num; j++ 
		<-sign
	

注意，这些发送表达式应该被放在它们的go函数体的最后面。对应的，我们还需要在main函数的最后从通道接收元素值，接收的次数也应该与手动启用的 goroutine 的数量保持一致。

其中有一个细节你需要注意。我在声明通道sign的时候是以chan struct作为其类型的。其中的类型字面量struct有些类似于空接口类型interface，它代表了既不包含任何字段也不拥有任何方法的空结构体类型。

注意，struct类型值的表示法只有一个，即：struct。并且，它占用的内存空间是0字节。确切地说，这个值在整个 Go 程序中永远都只会存在一份。虽然我们可以无数次地使用这个值字面量，但是用到的却都是同一个值。

当我们仅仅把通道当作传递某种简单信号的介质的时候，用struct作为其元素类型是再好不过的了。

再说回当下的问题，有没有比使用通道更好的方法？如果你知道标准库中的代码包sync的话，那么可能会想到sync.WaitGroup类型。没错，这是一个更好的答案。不过具体的使用方式我在后边讲sync包的时候再说。

3.3 怎样让我们启用的多个 goroutine 按照既定的顺序运行？

在很多时候，当我沿着上面的主问题以及第一个扩展问题一路问下来的时候，应聘者往往会被这第二个扩展问题难住。

所以基于上面问题中的代码，怎样做到让从0到9这几个整数按照自然数的顺序打印出来？你可能会说，我不用 goroutine 不就可以了嘛。没错，这样是可以，但是如果我不考虑这样做呢。你应该怎么解决这个问题？

先看答案

func main() 
	var count uint32
	trigger := func(i uint32, fn func()) 
		for 
			if n := atomic.LoadUint32(&count); n == i 
				fn()
				atomic.AddUint32(&count, 1)
				break
			
		
	

	for i := uint32(0); i < 10; i++ 
		go func(i uint32) 
			fn := func()  fmt.Println(i) 
			trigger(i, fn)
		(i)
	
	trigger(10, func() )

下面我们来分析实现过程，首先，我们需要稍微改造一下for语句中的那个go函数，要让它接受一个int类型的参数，并在调用它的时候把变量i的值传进去。为了不改动这个go函数中的其他代码，我们可以把它的这个参数也命名为i。

for i := 0; i < 10; i++ 
  go func(i int) 
    fmt.Println(i)
  (i)

只有这样，Go 语言才能保证每个 goroutine 都可以拿到一个唯一的整数。其原因与go函数的执行时机有关。

在go语句被执行时，我们传给go函数的参数i会先被求值，如此就得到了当次迭代的序号。之后，无论go函数会在什么时候执行，这个参数值都不会变。也就是说，go函数中调用的fmt.Println函数打印的一定会是那个当次迭代的序号。

然后，我们在着手改造for语句中的go函数。

for i := uint32(0); i < 10; i++ 
  go func(i uint32) 
    fn := func() 
      fmt.Println(i)
    
    trigger(i, fn)
  (i)

我在go函数中先声明了一个匿名的函数，并把它赋给了变量fn。这个匿名函数做的事情很简单，只是调用fmt.Println函数以打印go函数的参数i的值。

在这之后，我调用了一个名叫trigger的函数，并把go函数的参数i和刚刚声明的变量fn作为参数传给了它。注意，for语句声明的局部变量i和go函数的参数i的类型都变了，都由int变为了uint32。至于为什么，我一会儿再说。

再来说trigger函数。该函数接受两个参数，一个是uint32类型的参数i, 另一个是func()类型的参数fn。你应该记得，func()代表的是既无参数声明也无结果声明的函数类型。

trigger := func(i uint32, fn func()) 
  for 
    if n := atomic.LoadUint32(&count); n == i 
      fn()
      atomic.AddUint32(&count, 1)
      break
    
    time.Sleep(time.Nanosecond)
  

trigger函数会不断地获取一个名叫count的变量的值，并判断该值是否与参数i的值相同。如果相同，那么就立即调用fn代表的函数，然后把count变量的值加1，最后显式地退出当前的循环。否则，我们就先让当前的 goroutine“睡眠”一个纳秒再进入下一个迭代。

注意，我操作变量count的时候使用的都是原子操作。这是由于trigger函数会被多个 goroutine 并发地调用，所以它用到的非本地变量count，就被多个用户级线程共用了。因此，对它的操作就产生了竞态条件（race condition），破坏了程序的并发安全性。

所以，我们总是应该对这样的操作加以保护，在sync/atomic包中声明了很多用于原子操作的函数。

另外，由于我选用的原子操作函数对被操作的数值的类型有约束，所以我才对count以及相关的变量和参数的类型进行了统一的变更（由int变为了uint32）。

纵观count变量、trigger函数以及改造后的for语句和go函数，我要做的是，让count变量成为一个信号，它的值总是下一个可以调用打印函数的go函数的序号。

这个序号其实就是启用 goroutine 时，那个当次迭代的序号。也正因为如此，go函数实际的执行顺序才会与go语句的执行顺序完全一致。此外，这里的trigger函数实现了一种自旋（spinning）。除非发现条件已满足，否则它会不断地进行检查。

最后要说的是，因为我依然想让主 goroutine 最后一个运行完毕，所以还需要加一行代码。不过既然有了trigger函数，我就没有再使用通道。

trigger(10, func())

调用trigger函数完全可以达到相同的效果。由于当所有我手动启用的 goroutine 都运行完毕之后，count的值一定会是10，所以我就把10作为了第一个参数值。又由于我并不想打印这个10，所以我把一个什么都不做的函数作为了第二个参数值。