并发编程是现代软件开发的核心挑战之一。不同编程语言采用了不同的并发模型来解决多任务执行的问题。本文将系统对比主流的三大并发模型——共享内存模型、Actor 模型和 CSP 模型，并深入剖析 Go 语言基于 CSP 的并发设计，包括其调度器原理、channel 语义以及常见并发模式。
并发模型的三大流派
共享内存模型
共享内存模型是 C、C++、Java 等传统语言采用的并发方式。它的核心思想是多个线程共享同一块内存地址空间，通过锁（mutex）、信号量（semaphore）、条件变量等同步原语来协调对共享数据的访问。
#include <pthread.h>
#include <stdio.h>
int counter = 0;
pthread_mutex_t lock;
void* increment(void* arg) {
for (int i = 0; i < 1000; i++) {
pthread_mutex_lock(&lock);
counter++;
pthread_mutex_unlock(&lock);
}
return NULL;
}
int main() {
pthread_t t1, t2;
pthread_mutex_init(&lock, NULL);
pthread_create(&t1, NULL, increment, NULL);
pthread_create(&t2, NULL, increment, NULL);
pthread_join(t1, NULL);
pthread_join(t2, NULL);
printf("Counter: %d\n", counter);
pthread_mutex_destroy(&lock);
return 0;
}
共享内存模型的优势在于数据共享直接、通信开销低。然而，它也带来了严峻的挑战：死锁、竞态条件、内存可见性问题、以及调试困难。开发者必须精确管理锁的获取和释放顺序，稍有不慎就会引入难以复现的 bug。
Actor 模型
Actor 模型由 Carl Hewitt 在 1973 年提出，在 Erlang、Akka（Scala/Java）和 Elixir 等语言中得到了广泛应用。在 Actor 模型中，每个 Actor 是一个独立的计算单元，拥有自己的私有状态，Actor 之间只能通过异步消息传递进行通信。
-module(counter).
-export([start/0, increment/1, get/1]).
start() ->
spawn(fun() -> loop(0) end).
increment(Pid) ->
Pid ! increment.
get(Pid) ->
Pid ! {get, self()},
receive
{value, V} -> V
end.
loop(Count) ->
receive
increment ->
loop(Count + 1);
{get, From} ->
From ! {value, Count},
loop(Count)
end.
Actor 模型的核心优势是天然消除了共享状态带来的竞态条件，因为数据永远不共享，只通过消息传递。此外，Actor 模型天然支持分布式系统，本地 Actor 和远程 Actor 的通信方式完全一致。缺点是消息传递的开销相对较高，且调试追踪消息流可能比追踪锁更困难。
CSP 模型
CSP（Communicating Sequential Processes，通信顺序进程）由 Tony Hoare 在 1978 年的同名论文中提出。Go 语言的并发设计深受 CSP 影响，其核心思想是通过 channel 进行通信，而不是通过共享内存。
CSP 与 Actor 模型的关键区别在于：
- CSP 的通信是同步的（除非使用缓冲 channel），而 Actor 是异步消息传递
- CSP 的 channel 是独立的实体，可以多个 goroutine 共享同一个 channel；Actor 的消息直接发送到目标 Actor
- CSP 更关注通信的同步性质，强调通信本身作为同步点
Go 语言的 CSP 实现
Goroutine：轻量级线程
Goroutine 是 Go 并发的执行单元，由 Go 运行时管理，而非操作系统直接调度。一个 goroutine 的初始栈空间仅为 2KB，并且可以动态伸缩（最大可达 1GB 左右）。这使得在一个 Go 程序中轻松启动数万甚至数十万个 goroutine 成为可能。
package main
import (
"fmt"
"sync"
"time"
)
func worker(id int, wg *sync.WaitGroup) {
defer wg.Done()
fmt.Printf("Worker %d starting\n", id)
time.Sleep(time.Second)
fmt.Printf("Worker %d done\n", id)
}
func main() {
var wg sync.WaitGroup
for i := 1; i <= 5; i++ {
wg.Add(1)
go worker(i, &wg)
}
wg.Wait()
fmt.Println("All workers completed")
}
创建 goroutine 只需要在函数调用前加上 go 关键字，语法极其简洁。但这种简洁背后隐藏着强大的调度机制。
G-M-P 调度器
Go 的调度器实现了 M:N 的线程模型，即 M 个 goroutine 运行在 N 个操作系统线程上。调度器的核心组件是：
- G（Goroutine）：代表一个 goroutine，包含栈信息、程序计数器、状态等
- M（Machine）：代表一个操作系统线程，是执行 goroutine 的实体
- P（Processor）：逻辑处理器，代表执行所需的资源，包括本地可运行 goroutine 队列
每个 P 维护一个本地队列（最多 256 个 G），还有一个全局队列供所有 P 共享。当某个 P 的本地队列空了，它会尝试从全局队列获取，或者从其他 P 的队列偷取一半的任务（work stealing）。这种设计最大限度地减少了锁竞争，提高了扩展性。
调度器会进行以下类型的切换：
- 主动调度：goroutine 执行完或调用
runtime.Gosched()主动让出 - 被动调度：goroutine 因 I/O 操作、channel 阻塞、锁等待等原因挂起
- 抢占式调度：Go 1.14 引入了基于信号的协作式抢占，避免单个 goroutine 长期占用 CPU
package main
import (
"fmt"
"runtime"
"time"
)
func main() {
// 设置最大并行度等于 CPU 核心数
fmt.Println("NumCPU:", runtime.NumCPU())
fmt.Println("GOMAXPROCS:", runtime.GOMAXPROCS(0))
// 启动大量 goroutine 观察调度效果
for i := 0; i < 10; i++ {
go func(n int) {
for {
// 模拟 CPU 密集型工作
_ = n * n
time.Sleep(time.Millisecond * 10)
}
}(i)
}
time.Sleep(time.Second)
fmt.Println("All goroutines scheduled")
}
Channel 的同步与异步语义
Channel 是 Go 中 goroutine 之间通信和同步的核心机制。Channel 的设计精妙之处在于：通信即同步——当两个 goroutine 在同一个 channel 上执行发送和接收时，它们之间建立了同步点。
无缓冲 Channel
无缓冲 channel 要求发送和接收必须同时发生，否则发送方会阻塞直到接收方就绪。
package main
import "fmt"
func main() {
ch := make(chan int) // 无缓冲 channel
go func() {
fmt.Println("Receiver ready, waiting for data")
v := <-ch
fmt.Println("Received:", v)
}()
fmt.Println("Sender about to send")
ch <- 42 // 阻塞直到接收方就绪
fmt.Println("Send complete")
}
有缓冲 Channel
有缓冲 channel 允许发送方在缓冲区未满时非阻塞地发送数据。
package main
import "fmt"
func main() {
ch := make(chan int, 3) // 缓冲区大小为 3
ch <- 1
ch <- 2
ch <- 3
// ch <- 4 // 这里会阻塞，因为缓冲区已满
fmt.Println(<-ch) // 1
fmt.Println(<-ch) // 2
fmt.Println(<-ch) // 3
}
理解 channel 的关闭语义也很重要：关闭一个 channel 后，接收方仍然可以读取缓冲区中剩余的数据，但不能再发送数据。从已关闭且无缓冲的 channel 接收会返回零值，配合多重赋值可以检测 channel 状态。
package main
import "fmt"
func main() {
ch := make(chan int, 2)
ch <- 1
ch <- 2
close(ch)
for {
v, ok := <-ch
if !ok {
fmt.Println("Channel closed, no more data")
break
}
fmt.Println("Received:", v)
}
}
Select 的非确定性选择
select 语句是 Go 并发编程的强大工具，它允许 goroutine 同时等待多个通信操作。当多个 case 同时就绪时，select 会随机选择一个执行，体现了 CSP 的非确定性本质。
package main
import (
"fmt"
"time"
)
func main() {
ch1 := make(chan string)
ch2 := make(chan string)
go func() {
time.Sleep(time.Second)
ch1 <- "from ch1"
}()
go func() {
time.Sleep(time.Second * 2)
ch2 <- "from ch2"
}()
for i := 0; i < 2; i++ {
select {
case msg1 := <-ch1:
fmt.Println(msg1)
case msg2 := <-ch2:
fmt.Println(msg2)
case <-time.After(time.Second * 3):
fmt.Println("timeout")
}
}
}
select 还可以配合 default 分支实现非阻塞通信，这是构建高性能并发系统的重要技巧。
经典并发模式
Fan-Out / Fan-In 模式
Fan-Out 是将一个任务分发给多个 worker 并行处理，Fan-In 是将多个 worker 的结果收集汇总。
package main
import (
"fmt"
"sync"
)
// producer 生成数据
func producer(nums ...int) <-chan int {
out := make(chan int)
go func() {
for _, n := range nums {
out <- n
}
close(out)
}()
return out
}
// square 计算平方（worker）
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
for n := range in {
out <- n * n
}
close(out)
}()
return out
}
// merge 将多个 channel 合并（fan-in）
func merge(channels ...<-chan int) <-chan int {
var wg sync.WaitGroup
out := make(chan int)
output := func(ch <-chan int) {
defer wg.Done()
for n := range ch {
out <- n
}
}
wg.Add(len(channels))
for _, ch := range channels {
go output(ch)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
func main() {
in := producer(1, 2, 3, 4, 5)
// fan-out: 启动多个 square worker
c1 := square(in)
c2 := square(in)
c3 := square(in)
// fan-in: 合并结果
for result := range merge(c1, c2, c3) {
fmt.Println(result)
}
}
Pipeline 模式
Pipeline 将数据处理流程拆分为多个阶段，每个阶段由独立的 goroutine 负责，数据通过 channel 在阶段之间流动。
package main
import "fmt"
func generator(nums ...int) <-chan int {
out := make(chan int)
go func() {
for _, n := range nums {
out <- n
}
close(out)
}()
return out
}
func filter(in <-chan int, divisibleBy int) <-chan int {
out := make(chan int)
go func() {
for n := range in {
if n%divisibleBy != 0 {
out <- n
}
}
close(out)
}()
return out
}
func double(in <-chan int) <-chan int {
out := make(chan int)
go func() {
for n := range in {
out <- n * 2
}
close(out)
}()
return out
}
func main() {
nums := generator(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
filtered := filter(nums, 3) // 过滤掉能被 3 整除的数
result := double(filtered) // 剩余数字翻倍
for v := range result {
fmt.Println(v)
}
}
Worker Pool 模式
Worker Pool 模式限制并发执行的数量，适用于批量任务处理场景。
package main
import (
"fmt"
"sync"
"time"
)
func workerPool(jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
defer wg.Done()
for job := range jobs {
// 模拟耗时工作
time.Sleep(time.Millisecond * 100)
results <- job * job
}
}
func main() {
jobs := make(chan int, 100)
results := make(chan int, 100)
var wg sync.WaitGroup
// 启动 3 个 worker
for i := 0; i < 3; i++ {
wg.Add(1)
go workerPool(jobs, results, &wg)
}
// 发送 10 个任务
go func() {
for i := 1; i <= 10; i++ {
jobs <- i
}
close(jobs)
}()
// 等待所有 worker 完成并关闭 results
go func() {
wg.Wait()
close(results)
}()
// 收集结果
for result := range results {
fmt.Println("Result:", result)
}
}
并发模型对比总结
| 特性 | 共享内存 | Actor 模型 | CSP (Go) |
|---|---|---|---|
| 通信方式 | 共享变量+锁 | 异步消息 | 同步/异步 channel |
| 数据共享 | 直接共享 | 完全不共享 | 通过通信传递所有权 |
| 同步机制 | 锁、信号量 | 消息邮箱 | channel 通信 |
| 死锁风险 | 高 | 低（有监督者机制） | 中（channel 使用不当） |
| 分布式支持 | 困难 | 原生支持 | 需额外库支持 |
| 调试难度 | 高 | 中 | 低 |
| 性能开销 | 锁竞争激烈时低 | 消息序列化开销 | channel 切换开销 |
| 代表语言 | C/C++/Java | Erlang/Scala | Go |
Go 并发最佳实践
使用 channel 控制并发数量
不要用共享变量+锁来协调 goroutine，优先考虑 channel 传递信号。
避免 Goroutine 泄漏
确保每个创建的 goroutine 都有退出路径，特别是在使用无缓冲 channel 时。
// 错误示例：如果 nobody receives from done, the goroutine leaks
done := make(chan bool)
go func() {
// ... work
done <- true // 如果接收方没有就绪，这里会永远阻塞
}()
// 正确做法
done := make(chan bool, 1) // 使用缓冲 channel
go func() {
// ... work
done <- true
}()
用 Context 管理生命周期
context.Context 是 Go 1.7 引入的标准方式来传递截止时间和取消信号。
package main
import (
"context"
"fmt"
"time"
)
func longRunningTask(ctx context.Context) error {
select {
case <-time.After(time.Second * 2):
fmt.Println("Task completed")
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err := longRunningTask(ctx); err != nil {
fmt.Println("Task failed:", err)
}
}
优先使用 WaitGroup 等待 goroutine
不要依赖 time.Sleep 来等待 goroutine 完成，使用 sync.WaitGroup 或 channel 显式同步。
小心闭包变量捕获
循环中启动 goroutine 时，要特别注意变量捕获问题。
// 错误示例
for i := 0; i < 10; i++ {
go func() {
fmt.Println(i) // 所有 goroutine 可能打印相同的值
}()
}
// 正确做法
for i := 0; i < 10; i++ {
go func(n int) {
fmt.Println(n)
}(i)
}
常见陷阱与排查
死锁
最常见的死锁原因是所有 goroutine 都在等待彼此。Go 运行时会检测全部 goroutine 都永久阻塞的情况并 panic。
Channel 发送给 nil channel
向 nil channel 发送或接收操作会永远阻塞，这种 bug 很难发现。
关闭已关闭的 channel
重复关闭 channel 会引发 panic，应该保证 channel 的关闭只发生一次。通常由发送方来关闭 channel，或者使用 sync.Once。
竞态条件
使用 go run -race 或 go test -race 可以检测数据竞态。强烈建议在测试阶段始终开启竞态检测器。
总结
Go 语言通过 CSP 模型提供了简洁而强大的并发编程能力。与共享内存模型相比，Go 的 channel 机制降低了心智负担；与 Actor 模型相比，Go 的同步通信语义更加直观可控。掌握 goroutine、channel 和 select 的使用，理解 G-M-P 调度器的基本原理，熟练运用 fan-out/fan-in、pipeline、worker pool 等经典模式，将使你在 Go 并发编程中游刃有余。记住 Go 的并发格言：不要通过共享内存来通信，而要通过通信来共享内存。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。