ConcurrentML：超越Go的并发模型

# ConcurrentML：改进 Go 的并发模型 来源：https://nikhilism.com/post/2026/golang-racket-concurrency/ - events：将通信操作作为值 (https://nikhilism.com/post/2026/golang-racket-concurrency/#events-communication-operations-as-values) - 封装事件 (https://nikhilism.com/post/2026/golang-racket-concurrency/#wrapping-events) - 组合选择 (https://nikhilism.com/post/2026/golang-racket-concurrency/#composing-choice) - 基于事件构建 (https://nikhilism.com/post/2026/golang-racket-concurrency/#building-on-events) - 守卫事件 (https://nikhilism.com/post/2026/golang-racket-concurrency/#guard-events) - 处理取消 (https://nikhilism.com/post/2026/golang-racket-concurrency/#handling-cancellation) - 打印服务中的取消 (https://nikhilism.com/post/2026/golang-racket-concurrency/#cancellation-in-the-print-service) - 收尾散落的线程 (https://nikhilism.com/post/2026/golang-racket-concurrency/#wrapping-up-loose-threads) - 关于 `close` 的思考 (https://nikhilism.com/post/2026/golang-racket-concurrency/#closeing-thoughts) - 参考文献 (https://nikhilism.com/post/2026/golang-racket-concurrency/#references) - 附录：超越通道的事件 (https://nikhilism.com/post/2026/golang-racket-concurrency/#appendix-events-beyond-channels) ***LLM 使用声明**：本文主要由人撰写。但我使用了 LLM 进行语音转文字、基础语法校正、写作反馈以及自动添加内联 API 文档链接。* Go 编程语言在将 CSP（Communicating Sequential Processes，通信顺序进程）并发模型推向广泛应用方面功不可没。在 CSP 模型中，通道通常是*无缓冲的*，它们不仅作为通信手段，还承担着同步的作用。也就是说，发送或接收会阻塞，直到另一端准备就绪，然后值被原子性地从一个线程转移到另一个线程。此外，[`select`](https://go.dev/ref/spec#Select_statements) 语句保证在其子句的所有通道操作中，只有一个会被选中，其余的不会被处理。这使得多种消息传递模式能够被简洁地表达。 CSP 是一个 [1970 年代](https://web.archive.org/web/20260922190027/https://www.cs.cmu.edu/~crary/819-f09/Hoare78.pdf)的古老思想，虽然 Go 相对忠实地遵循了该模型，但其他语言将其推进了一步。在我看来，最有意思的是 Concurrent ML（CML）。CML 于 1990 年代中期引入，至今仍有许多值得现代语言设计者学习的地方。在本文中，我想论证 CML 在 CSP 之上的演进为实现并发系统提供了引人注目的优势，修复了 Go 的若干缺陷。 **CML 的核心思想是将通信操作从可执行的语句提升为描述该操作的*值*。该描述的执行可以单独进行。** 如果这一点一开始不太清楚，请别担心，整篇文章都是对这个思想的阐述。它解锁了若干相对于 Go/CSP 的强大改进： 1. 它将选择（`select` 所做的事）和取消统一为一个通用的抽象。 2. 这个抽象使得并发的组合和封装成为可能，而这在 Go 中是不可能的。 除此之外，CML 还通过直接的垃圾回收修复了 Go 最大的并发陷阱之一——goroutine 如果没有正确关闭就会泄漏。 虽然这些思想来自 CML，但今天 [Racket](https://racket-lang.org/) 是最能体现这些思想的编程语言。Racket 正在积极开发中，并且在其中全面支持并使用 CML 模型。基于此，我将使用 Racket 演示所有示例，并根据上下文交替使用 CML 或 Racket 这两个术语。同样，我将交替使用 *goroutine* 和 *线程* 这两个词。它们的语义是相同的，至少就本文的目的而言是这样。 本文没有提出任何新东西。它的目的是向从业者介绍一个基本停留在学术界的模型。 本文假设你对 Go 语法和通道的行为有合理的了解。不假设你直接熟悉 Racket，但你应该能跟上类 Scheme 语言中关于函数调用和递归的基本概念。 要尝试 Racket 代码片段，请[下载 Racket](https://download.racket-lang.org/)，启动 DrRacket 编辑器，将示例粘贴到主窗口中，然后按 Run 按钮执行。 ## events：将通信操作作为值 在 Go 中，如果 `ch` 是一个通道，以下两个操作用于在通道上发送或接收消息。 ```racket ; 不要运行这个，会死锁 ch := make(chan int) ch <- 5 x := <-ch ``` 这些通道操作语句在被执行到时[立即生效](https://go.dev/tour/concurrency/2)。当与 [`select`](https://go.dev/ref/spec#Select_statements) 语句配合使用时，这些表达式可以在 `case` 子句中组合，以表达运行时可以在其之间进行选择的潜在选项。 当执行到达 [`select`](https://go.dev/ref/spec#Select_statements) 语句时，所有通道操作都会被轮询，以查看哪些操作已准备好执行。从已就绪的操作中选出一个。可以保证在本次执行中不会有其他通道操作完成。 ```go select { case ch1 <- 5: case y := <- ch2: default: // 如果其他操作都没有就绪，则选择此分支。 } ``` 在 select 内部，通道操作充当*描述*，`select` 语句使用这些描述来确定要轮询哪些通道以及执行什么操作。也就是说，命中第一个 case 并不会立即阻塞整个 `select`。 然而在 Go 中，这种描述形式无法与 [`select`](https://go.dev/ref/spec#Select_statements) 语句解耦并独立存在。一旦你尝试这样做，它就变成了前面的例子，语句会尝试立即执行。这阻碍了组合和封装。函数无法返回一个你可以在之后选择 `select` 的描述。你也不能将描述存储在结构体中随身携带。 Racket 中的通道同样起步，并提供了阻塞式发送和接收操作的直接对应物。 ```racket ; ch := make(chan int) (define ch (make-channel)) ; ch <- 5 (channel-put ch 5) ; x := <-ch (define x (channel-get ch)) ``` 然而，它们还提供了一种独立的方式来描述操作。 ```racket ; 类似于 `case ch1 <- 5`。 (channel-put-evt ch 5) ; 没有 channel-get-evt。 ; 通道本身可以直接用作从它接收的描述。 ``` **这是一个返回值的表达式，*不是*一个动作语句！** `channel-put-evt` 返回一个在 CML 中称为*事件*的值。它不会立即将值 5 发送到通道上。该事件可以在之后被执行以实际施加其效果。 因为它是一个值，就像数字 5 一样，我们可以对它做很多有用的事情，这在 Go 中是不可能的： - 将其赋值给变量， - 从函数返回它， - 将其放入结构体中， - 将我们正在执行通道操作这一事实抽象到普通函数后面， - 用额外的并发组合子封装它以表达模式 此外，事件比通道操作更具通用性，能够表达其他语言中"事件循环"通常暴露的 I/O 操作。 事件的实际效果由 [`sync`](https://docs.racket-lang.org/reference/sync.html#%28def._%28%28quote._~23~25kernel%29._sync%29%29) 函数触发 ```racket (sync (channel-put-evt ch 5)) ``` 使用 [`sync`](https://docs.racket-lang.org/reference/sync.html#%28def._%28%28quote._~23~25kernel%29._sync%29%29) *尝试同步*一组事件。就像 `select` 一样，传递给它的所有事件都会被轮询。一个*就绪*的事件会被选中，`sync` 返回被选中事件的*同步结果*，这是每个事件可以自行定义的。对于通道上的发送操作，该结果就是事件本身。对于从通道接收，则是接收到的值。当多个事件就绪时，会随机选择一个，与 Go 相同。 例如，这里有一个 Racket 线程，它计算一个数字列表的和并通过通道返回响应。 ```racket #lang racket/base (define (sum items resp-ch) (define total (apply + items)) (channel-put resp-ch total)) (define ch (make-channel)) (define the-numbers (list 7 2 8 -9 4 0)) (thread (λ () (sum the-numbers ch))) (define total (sync ch)) (printf "Sum of ~a is ~a" the-numbers total) ``` 注意，这种描述与执行的分离可以很容易地重新组合。[`channel-put`](https://docs.racket-lang.org/reference/channel.html#%28def._%28%28lib._racket%2Fprivate%2Fmisc..rkt%29._channel-put%29%29) 可以被定义为： ```racket (define (channel-put ch val) (sync (channel-put-evt ch val))) ``` ### 封装事件 > 在整篇文章中，我们将大量使用匿名函数。在 Go 中，它们通过 `func` 创建。在 Racket 中，使用 [`lambda`](https://docs.racket-lang.org/reference/lambda.html#%28form._%28%28lib._racket%2Fprivate%2Fbase..rkt%29._lambda%29%29) 形式，或字面量 [`λ`](https://docs.racket-lang.org/reference/lambda.html#%28form._%28%28lib._racket%2Fprivate%2Fbase..rkt%29._lambda%29%29)。如果你想复习这些知识，请参阅 Go 的[这里](https://go.dev/doc/codewalk/functions/)或 Racket 的[这里](https://docs.racket-lang.org/guide/syntax-overview.html#%28part._.Anonymous_.Functions_with_lambda%29)。 [Tour of Go](https://go.dev/tour/concurrency/5) 中 `select` 语句的第一个示例使用一个 goroutine 来计算斐波那契数列，并通过通道传递响应。它还使用了一个额外的通道来告诉 goroutine 退出。以下是 Racket 翻译版本： ```racket #lang racket/base (define (fibonacci out-ch quit-ch) (let loop ([x 0] [y 1]) (sync (handle-evt (channel-put-evt out-ch x) (λ (_) (loop y (+ x y)))) (handle-evt quit-ch (λ (_) (displayln "quit")))))) (module+ main (define c (make-channel)) (define quit-ch (make-channel)) (thread (λ () (for ([i (in-range 10)]) (displayln (channel-get c))) (channel-put quit-ch #f))) (fibonacci c quit-ch)) ``` 这里有一些函数式语言特有的形式，比如用命名 `let` 递归地表达 Go 的 `for {}` 无限循环。`in-range` 是 Racket 的数字迭代形式，所以 `for` 循环执行 10 次，获取前 10 个斐波那契数。 就像 [`select`](https://go.dev/ref/spec#Select_statements) 接受多个通道操作一样，[`sync`](https://docs.racket-lang.org/reference/sync.html#%28def._%28%28quote._~23~25kernel%29._sync%29%29) 接受多个事件，然后阻塞直到至少一个事件准备就绪可以同步。 这个第一个示例引入了一个新函数 [`handle-evt`](https://docs.racket-lang.org/reference/sync.html#%28def._%28%28quote._~23~25kernel%29._handle-evt%29%29)。这是一个*事件组合子*——一个增强事件的函数，它本身返回一个事件。 在 Go 的 [`select`](https://go.dev/ref/spec#Select_statements) 中，一个 `case` 子句后面可以跟一系列语句，当且仅当该子句被选中且操作被执行时，这些语句才会被执行。你可以将 `select` 的子句视为*提交点*，特定的通信操作在这些点上被提交，而其他操作被放弃。有了这个心智模型，我们可以在提交点*之前*（*准备*）或提交点*之后*（*后提交*）做事情。 一旦我们开始组合事件，将准备和后提交操作与 [`sync`](https://docs.racket-lang.org/reference/sync.html#%28def._%28%28quote._~23~25kernel%29._sync%29%29) 关联就不再有意义了，否则，每当你从函数返回一个事件时，还需要单独返回准备操作。 相反，我们可以将这些步骤附加到事件本身。这通过一个包装函数来完成，该函数接受一个事件和一个函数。每个包装器以特定方式增强事件。这些也称为*事件组合子*。 我们的第一个包装器是 [`handle-evt`](https://docs.racket-lang.org/reference/sync.html#%28def._%28%28quote._~23~25kernel%29._handle-evt%29%29)，它将*后提交*操作与一个事件关联。它返回一个新的、包装后的事件。当被包装的事件被 [`sync`](https://docs.racket-lang.org/reference/sync.html#%28def._%28%28quote._~23~25kernel%29._sync%29%29) 选中时（即赢得竞争），关联的函数会以事件的同步结果作为参数被执行。在上面的示例中，我们使用这些操作来决定是继续循环还是退出。 注意，这些包装器利用了事件的通用性质，将许多常见操作纳入该框架。所有包装器都在封装其输入事件、增强它们并返回输出事件。这些事件的使用者*不能*也*不需要*弄清楚事件背后隐藏了什么。 作为封装的另一个示例，退出*模式*可以被提取到一个 `quit-evt` 中。实现之外的任何部分都不需要知道退出实际上是通过通道实现的。 ```racket #lang racket/base ; 返回一个 quit-evt，以及一个触发退出的函数。 (define (make-quit-evt) (define my-quit-ch (make-channel)) ; 在这种情况下，我们的 quit-evt 只是一个通道，但没有人需要知道这一点。 ; values 是 Racket 中函数返回多个值的方式。 (values my-quit-ch (λ () (channel-put my-quit-ch #f)))) (define (fibonacci out-ch quit-evt) (let loop ([x 0] [y 1]) ; 之后可以在这里使用。 (sync (handle-evt (channel-put-evt out-ch x) (λ (_) (loop y (+ x y)))) quit-evt))) (module+ main (define c (make-channel)) ; define-values 是对返回多个值的函数进行"解包"的操作。 ; 类似于 Go 中 result, err := Func() 的用法。 (define-values (quit-evt quit-fn) (make-quit-evt)) (thread (λ () (for ([i (in-range 10)]) (displayln (channel-get c))) (quit-fn))) (fibonacci c quit-evt)) ``` 我希望现在已经清楚了，将通道操作作为描述性值来处理解锁了组合能力！ ### 组合选择 能够返回事件意味着 Racket 代码经常从各种来源构建事件列表，然后对所有事件调用 [`sync`](https://docs.racket-lang.org/reference/sync.html#%28def._%28%28quote._~23~25kernel%29._sync%29%29)。`sync` 实际上是在这些事件之间执行一个*选择*操作，Racket 允许该概念也被分离为一个包装事件。组合子是 [`choice-evt`](https://docs.racket-lang.org/reference/sync.html#%28def._%28%28quote._~23~25kernel%29._choice-evt%29%29)。 这是一个常见的模式："从 2 个不同来源获取数据，接受最快返回的那个"。我们可以轻松地将其与超时组合。视野内完全不需要 [`context.Context`](https://pkg.go.dev/context#Context)！ ```racket #lang racket/base ; 这个 timeout-evt 会在本文其余部分使用。 ; 它将相对时间转换为绝对时间，并将其与内置的 alarm-evt 一起使用。 (define (timeout-evt msec) (alarm-evt (+ (current-inexact-monotonic-milliseconds) msec) #t)) ; API 是虚构的 (define (http-get-evt url) ...) (define (fetch-from-fastest-url url1 url2) (choice-evt (http-get-evt url1) (http-get-evt url2))) ; 执行 2 个 http 请求，返回最快响应者的结果，同时用超时进行限制。 (sync (fetch-from-fastest-url url1 url2) (timeout-evt 5)) ``` 这种组合在 Go 中是*不可能*的——