[publish] /article/clj7

1 files changed, 225 insertions, 24 deletions

diff --git a/content/article/clj7.md b/content/article/clj7.md
index 500900ba..d23dfc00 100644
--- a/content/article/clj7.md
+++ b/content/article/clj7.md
@@ -1,12 +1,13 @@
 +++
 title = 'Coffee Break Clojure, Async'
 subtitle = '异步编程和并发控制'
-slug = ''
-tags = []
+slug = 'clj7'
+tags = ["Go","Clojure","Lisp","编程"]
 categories = '茶歇蔻芥'
-date = 2026-10-11T18:35:50+08:00
-draft = true
+date = 2026-10-12 23:48:37
+draft = false
 toc = true
+fedi = 'https://akk.eltr.ac/objects/545d282b-4db6-4af9-a2e7-4440dd8ec30b'
 +++
 
 [上一篇](/article/clj6/)我提及了如何用 [Ring](https://github.com/ring-clojure/ring) 开发 HTTP 服务器，也提到 Ring 提供了同步 API 和异步 API。那么，什么是异步编程？为什么需要异步？在什么情况下需要让一个程序同时做很多事情？<!--more-->
@@ -15,7 +16,7 @@ toc = true
 
 同步（synchronization）的中文名字可能会让人误以为是多个实体同时进行操作，实际上同步的意思是指系统中所有事件都依照相同的顺序进行协调。可以用工厂流水线举例，假如流水线中一个工人需要等上一个工人交付一部分工作之后才能干活，整个流水线都要所有人一起干活才能转动起来，这个流水线就是同步的。反之，假如某些工作可以分开做，比如一部分人负责生产 A 零件，另一部分人负责生产 B 零件，两者互不干扰，而最终把 A 和 B 组装在一起的又是另一拨人，这个系统就是异步的（asynchronous）。
 
-异步（asynchrony）和并发（concurrency）很容易搞混，并不是同一个概念。异步是指系统中各个事件的状态不同步，而并发通常是指多个线程（或类似线程的东西，比如 `goroutine`）同时做类似的操作。
+异步（asynchrony）和并发（concurrency）很容易搞混，并不是同一个概念。异步是指系统中各个事件的状态不同步，而并发通常是指多个线程（或类似线程的东西）同时做类似的操作。
 
 以下分别是异步和并发的例子：
 
@@ -65,7 +66,7 @@ toc = true
 
 把 `client/get` 包裹在 `future` 宏里之后，我们立刻得到了 `response`，这个 `response` 并不是实际的 HTTP 响应，而是对未来能够获取到的数据的引用。
 
-立刻拿到这个引用之后，我们立刻处理手头已有的数据。等这些数据处理完成之后，尝试 `deref`（解引用）一开始拿到的引用，此时 HTTP 请求已经发出有一段时间了，我们有可能直接就能拿到数据，然后就可以开始处理了；如果没拿到，也只需要再等一会儿，比一开始就等着什么也不做强多了。
+拿到这个引用之后，我们立刻处理手头已有的数据。等这些数据处理完成之后，尝试 `deref`（解引用）一开始拿到的引用，此时 HTTP 请求已经发出有一段时间了，我们有可能直接就能拿到数据，然后就可以开始处理了；如果没拿到，也只需要再等一会儿，比一开始就等着什么也不做强多了。
 
 持有对未来可用数据的引用，和持有对某一块内存区域的“引用”，有什么区别呢？
 
@@ -73,7 +74,7 @@ toc = true
 
 以这个 Go 程序为反例：
 
-```javascript
+```go
 package main
 
 import (
@@ -107,6 +108,28 @@ Go 语言也有自己的异步编程范式，实际上不必这么写。之后�
 
 如果你熟悉 JavaScript 的 `async` 和 `await`，那 Clojure 的 `future` 和 `deref` 就很容易理解。
 
+此外，`(deref ref)` 可以简写为 `@ref`。不过用 `@` 语法糖的话就用不了 `deref` 的超时控制。`(deref ref)` 默认会一直等待，如果 `ref` 的值一直没来，程序就会卡死；可以用 `(deref ref timeout fallback)` 做超时控制，如果经过了 `timeout` 毫秒还没能解引用成功，就用 `fallback` 值。
+
+在继续之前我还要提醒对异步编程不那么熟悉的读者，无论是哪种异步机制，无论怎么实现并发，线程的执行顺序都是不固定的。
+
+```clojure 
+(dotimes [number 10] (future (println number)))
+;; 05
+;; 423
+;;
+;; 8
+;; 9
+;; nil1
+;; 7
+;;
+;;
+;; 6
+```
+
+上面的输出很乱，原因在于 `println` 的**换行符**何时打印也不能确定，`dotimes` 本身的返回值 `nil` 也混在这些换行得乱七八糟的数字当中。
+
+有办法能控制并发的执行顺序，但要保证顺序必然要有等待、排队的过程，速度就有损失，甚至有可能变成串行的。如果要严格同步，那就不应该异步——是的，这是句废话。总之，需要严格保证执行顺序的操作不该异步。
+
 ## 不阻塞的异步 
 
 尝试解引用 `future` 返回的引用可能会阻塞当前线程，如果希望当前线程不被阻塞呢？有时候我们并不需要严格同步数据，如果新值没有到达，那就用旧的值。
@@ -119,7 +142,7 @@ Go 语言也有自己的异步编程范式，实际上不必这么写。之后�
 (def counter (agent 0))
 ;; => #'user/counter
 
-(deref counter)
+@counter
 ;; => 0 
 
 (future 
@@ -128,20 +151,22 @@ Go 语言也有自己的异步编程范式，实际上不必这么写。之后�
     (send counter inc)))
 ;; => #object[clojure.core$future_call$reify__8574 0x4abf3f0 {:status :pending, :val nil}]
 
-(deref counter)
+@counter
 ;; => 0 
 
 ;; 五秒过后 
-(deref counter)
+@counter
 ;; => 1
 ```
 
-`doseq` 函数接收类似 `let` 的绑定，用于遍历序列。`(doseq [item data] (f item))` 相当于 `(doall (map #(f %) data))`——`doall` 立刻执行惰性序列中所有函数的副作用，而 `doseq` 则本来就是用于处理副作用的，不创建函数。
+`doseq` 宏接收类似 `let` 的绑定，用于遍历序列。`(doseq [item data] (f item))` 相当于 `(doall (map #(f %) data))`——如果不需要操作集合数据，比起 `map` 和 `doall`，`doseq` 宏看起来更干净。
 
-上述代码中传入的是无穷的惰性序列，所以会无限循环。每次循环都让线程先睡五秒钟，然后再执行操作。
+上述代码中传入的是无穷的惰性序列，所以会无限循环。每次循环都让线程先睡五秒钟，然后再执行操作。放在 `future` 里是为了不阻塞主线程。
 
 这里的 `send` 函数是专门和 `agent` 一起用的，作用是==把一个函数发送给 `agent`，让 `agent` 对自己保存的数据做处理==。这个过程是异步的，`send` 会立刻返回 `agent`，之后再解引用 `agent` 就能拿到 `agent` 当前的值。
 
+`agent` 是线程安全的，不必手动加锁。
+
 你可能发现了，这难道不是非常**不**函数式吗？比方说，如果不把 `agent` 用于异步编程，我们完全可以违反「不允许可变状态」的规则，做这种事情：
 
 ```clojure 
@@ -186,7 +211,7 @@ Go 语言也有自己的异步编程范式，实际上不必这么写。之后�
   ;; 后续传入的可变数量的参数（如这里的 url），会变成 crawl-append 剩下的参数
   (send crawler crawl-append url))
 
-(deref crawler)
+@crawler
 ;; 会返回由当前已经爬取完成的 HTML 字符串组成的向量
 ```
 
@@ -237,7 +262,7 @@ org.clojure/core.async {:mvn/version "1.9.865"}
 [org.clojure/core.async "1.9.865"]
 ```
 
-`core.async` 几乎和 Go 语言异步编程的要素一模一样，尤其是 `go` block——把语句放在 `go` block 里面，语句就会异步执行。把消息队列 `chan` 当作一等公民的做法，也如出一辙。根据 [Clojure 官方的博客文章](https://clojure.org/news/2013/06/28/clojure-clore-async-channels)，这种异步编程风格其实一段历史了，Go 也是借鉴并现代化了这种做法。
+`core.async` 几乎和 Go 语言异步编程的要素一模一样，尤其是 `go` block——把语句放在 `go` block 里面，语句就会异步执行。把消息队列 `chan` 当作一等公民的做法，也如出一辙。根据 [Clojure 官方的博客文章](https://clojure.org/news/2013/06/28/clojure-clore-async-channels)，这种异步编程风格其实有一段历史了，Go 也是借鉴并现代化了这种做法。
 
 > The roots of this style go back at least as far as Hoare’s Communicating Sequential Processes (CSP), followed by realizations and extensions in e.g. occam, Java CSP and the Go programming language.
 
@@ -249,29 +274,205 @@ org.clojure/core.async {:mvn/version "1.9.865"}
 (def queue (chan))
 
 (go 
-  (<! queue "Hello?"))
+  (>! queue "Hello?"))
 
 (go 
-  (println (>! queue)))
+  (println (<! queue)))
 ```
 
+上述例子中，我们启动了两个 `go` 线程，一个线程把 `"Hello"` 放进 `queue` 里，另一个线程把 `"Hello"` 取出来。
+
 `chan` 函数用于创建 *channel*（频道），可以理解为消息队列。对于 *channel*，有放（put）和拿（take）的操作，分别用 `>!`（放）和 `<!`（拿）表示。这里的 `!` 类似于 `?`，是 Clojure 的命名习惯，用于标识会对状态进行变更的操作。说到底，我们的确对状态进行了变更，上述语句缺乏引用透明性，不够函数式。
 
-`go` 在 Clojure 里是个宏，在 Go 里是个放在函数调用之前的关键字，效果都是让语句异步执行。值得注意的是，Clojure 中所有形式都是有返回值的，`go` 会返回一个 *channel*，`go` block 里的语句执行完成后，会把返回值通过 *channel* 传递回来。
+`chan` 可以设置缓冲区大小，`(chan 5)` 就是缓冲区大小为 `5` 的队列，一旦队列满了，再试图往里面放消息，放消息的线程就会被阻塞。
 
-上述例子中，我们启动了两个 `go` 线程，一个线程把 `"Hello"` 放进 `queue` 里，另一个线程把 `"Hello"` 取出来。
+```clojure 
+(def only-put-once (chan 1))
 
-放和拿的操作可以阻断线程，因此可以用作**并发控制**。
+(go (>! only-put-once 1)
+    ;; 下面这个操作会阻塞这个 go 线程
+    (>! only-put-once 1))
+```
 
-学过操作系统的读者可能知道信号量（semaphore）这个东西，我们可以用 `chan` 实现简单的信号量。比方说，我们先给一个 `chan` 存入 50 个信号，然后启动不定数量的 `go` 线程，但在启动线程之前，派发器必须要拿到信号量才行。如果信号量不够（尝试从空队列里取数据），就会阻塞派发器所在的线程，直到有信号量被回收。
+反之，如果队列空了还试图从里面拿东西，就会阻塞拿消息的线程。
 
 ```clojure 
-(def semaphore (chan))
+(def empty-chan (chan))
+
+(go (<! empty-chan)
+    ;; 除非有人往 empty-chan 里放东西
+    ;; 否则下面这句话不会被打印
+    (println "Help! I'm blocked!"))
+```
 
+`go` 在 Clojure 里是个宏，在 Go 里是个放在函数调用之前的关键字，效果都是让语句异步执行。不过 Clojure 中的 `go` 会返回一个 *channel*，`go` block 里的语句执行完成后，会把返回值通过 *channel* 传递回来。[^2]
+
+放和拿的操作可以阻塞线程，因此可以用作**并发控制**。
+
+学过操作系统的读者可能知道信号量（semaphore），我们可以用 `chan` 实现简单的信号量。比方说，我们先给一个 `chan` 存入 50 个信号，然后启动不定数量的 `go` 线程，但在启动线程之前，派发器必须要拿到信号量才行。如果信号量不够（尝试从空队列里取数据），就会阻塞派发器所在的线程，直到有信号量被回收。
+
+```clojure 
+(require '[clojure.core.async :refer [go chan put! <!! >!]])
+;; 定义信号量
+(def semaphore (chan))
+;; 将 50 个空 map 放入信号量中
 (dotimes [_ 50]
-  (>!! semaphore true))
+  (put! semaphore {}))
+;; 启动一个无限循环派发器
+(doseq [number (range)]
+  ;; 尝试获取信号量，如果已经消耗完 50 个，阻塞派发器
+  (<!! semaphore)
+  ;; 获取到信号量时，启动 go 线程
+  (go (Thread/sleep (rand-int 10000))
+      (println number)
+      ;; 线程工作完成后，把一个信号放入 semaphore
+      (>! semaphore {})))
 ```
 
-## Footnotes
+上述程序永远不会同时运行超过 50 个 `go` 线程。也可以反着实现，定义大小为 50 的 *channel*，每派发一个线程之前都要放个东西到 *channel* 里，比如 `{}`[^3]，放满了话的话就阻塞。不妨试着在 REPL 里写一写试试？
+
+需要注意的是放和拿的操作有很多变体，在不同场景下使用。之前提到的 `>!` 和 `<!` 只能放在 `go` block 中，放在其他地方会报错。
+
+在 `go` block 外，可以用 `>!!` 和 `<!!`，一般不建议用，因为会阻塞主线程。更好的选择是 `put!`（放）和 `take!`（拿），这两个操作是异步的，无论放没放进去、拿没拿出来，都不会阻塞当前的线程（但是会阻塞异步的子线程）。
+
+由于 `take!` 是异步的，不会立刻返回值，所以得传一个回调函数给它，这样在能拿到值的时候，就会调用回调函数。下面是 [ClojureDocs](https://clojuredocs.org/clojure.core.async/take!#example-542c8291e4b05f4d257a2974) 的示例：
+
+```clojure 
+user=> (def c (chan 1))
+#'user/c
+
+user=> (take! c
+              (fn [x]
+                (println "Clojure callback value " x)))
+nil
+
+user=> (put! c "XYZ")
+Clojure callback value  XYZ
+true
+
+user=> (put! c "XYZ")
+true
+```
+
+此外还有 `offer!`（放；给）和 `poll!`（拿；查），这两个操作是同步的，但也不会阻塞线程。`offer!` 会检查  *channel* 的缓冲区是不是满的，如果满了就返回 `nil`，如果能放进去就返回 `true`；`poll!` 也类似，能拿就拿，不能拿就返回 `nil`。
+
+## 像 Go 程序员一样写 Clojure
+
+Go 语言程序经常用 *goroutine*[^4] 和 *channel* 实现各种并发处理管线。我刚才已经演示了两个线程（协程）用 *channel* 异步通信，演示了派发器和并发控制。我们还可以把更多 Go 语言的编程模式搬到 Clojure 里来。
+
+我前面提到，`core.async` 提供的这套异步编程模式很**不**函数式，因为 *channel* 就是存放状态的队列，与之相关的放和拿的操作也不具有纯函数应有的引用透明性。
+
+不过这是否值得呢？接下来我要用 Go By Example[^5] 网站上的 [Stateful Goroutines](https://gobyexample.com/stateful-goroutines) 的例子回答这个问题。
+
+```go
+    reads := make(chan readOp)
+    writes := make(chan writeOp)
+
+    go func() {
+        var state = make(map[int]int)
+        for {
+            select {
+            case read := <-reads:
+                read.resp <- state[read.key]
+            case write := <-writes:
+                state[write.key] = write.val
+                write.resp <- true
+            }
+        }
+    }()
+```
+
+`select` 经常和 `chan` 一起用，放在 `select` 的哪个 `case` 中的 `chan` 先来消息，就先执行哪一个 `case`，也就是说可以同时等待多个 `chan`。
+
+上述代码中的 *goroutine* 管理了一个状态 `state`，即 *stateful goroutine*。它接收其他 *goroutine* 通过 *channel* 传递过来的消息，然后通过消息里携带的 `resp`（也是个 *channel*）把处理结果写回那些 *goroutine*。
+
+简单来说，这个 *stateful goroutine* 接收读操作和写操作。读者和写者不持有状态，只是发消息给 *stateful goroutine*，让它帮忙管理状态，然后给它一个 *channel*，稍后把处理结果发送回去。
+
+```go 
+    for range 100 {
+        go func() {
+            for {
+                read := readOp{
+                    key:  rand.Intn(5),
+                    resp: make(chan int)}
+                reads <- read
+                <-read.resp
+                atomic.AddUint64(&readOps, 1)
+                time.Sleep(time.Millisecond)
+            }
+        }()
+    }
+```
+
+上述代码派发了 100 个读者 *goroutine*，每个 *goroutine* 都把封装好的读操作写入 `reads` *channel*，然后等待 `resp` 被发回来。相对应地，还有写者 *goroutine*，这里就略过了。
+
+不知道你有没有注意到几件事：
+
+1. 管理状态的 *goroutine*、读者和写者之间，是**解耦**的，我们只需要有一个中间人（比方说 `main` 组件）把 *channel* 发给他们，他们就可以在没有代码依赖的情况下相互通信
+    - 这很方便测试，假设我要测试 *Stateful goroutine* 是否工作正常，我不需要启动读者和写者，我只需要把测试数据写入 `reads` 和 `writes`。
+    - 这让架构变得灵活，我可以随意重构三个组件，完全不会互相影响，不会出现某个组件的函数签名改了，另一个组件的函数调用也要改的情况。
+2. 我们把状态管理集中到了一个组件里，读者和写者可以是无状态的。
+    - 函数式编程的重点就是要管理状态和副作用，用 *goroutines* 和 *channel* 把状态管理集中到一个组件，也算是殊途同归了。
+3. 三个组件在共享数据，但是完全不需要加锁！
+
+要在 Clojure 里实现类似 `select` 的结构，可以用到 `alt!`（*alternative*）。
+
+```clojure
+;; (go-loop ...) = (go (loop ...))
+(go-loop [state {:key "val"}]
+  (alt!
+    reads ([read-op] 
+            (>! (:chan read-op) ((:key read-op) state))
+            (recur state))
+    writes ([write-op]
+             (recur (assoc state (:key write-op) (:val write-op))))))
+```
+
+不过，如你所见，除了 `>!`，这比 Go 语言的版本要更函数式一些，因为我们没有给 `state` 赋值，而是在 `recur` 一次又一次的递归调用中，不断给 `go-loop` 传入了不同的参数而已。
+
+我不久前就在工作中解决了一个不大不小的性能问题：系统中有大量 *goroutine* 等待 SQLite 的锁，导致计算资源闲置，而要写入 SQLite 的数据不过是些日志，并不着急同步。于是，我的解决方式是引入一个写者，所有写入数据库的操作都通过 *channel* 传递给这个写者，由写者这一个实体写数据，这样就不需要锁了。其他 *goroutine* 把写操作的情况放入 *channel* 之后就可以继续做自己的事情，不必等着把日志写进数据库。[^6]
+
+把副作用、状态管理和 I/O 操作集中在系统的一个组件，通过 *channel* 解耦其他组件与状态管理组件，能很好地降低系统的复杂性。
+
+## 我应该用什么？
+
+`future` 和 `deref` 很简单，`agent` 和 `send` 似乎也不错，`go` 和 `chan` 更是能很好地管控系统的复杂性，我究竟应该选哪个？
+
+老实说，我觉得这是程序员需要多写代码才能建立起来的认识。读者应该在实际的场合下权衡取舍，而不是选定一个万金油方案。这些方案都有利弊。
+
+`future` 和 `deref` 基本对应 JavaScript 的 `async` 和 `await`——**轻量**，易用，非常函数式。可如果要构建复杂系统，严格控制并发，甚至根据系统负载动态调整并发度，这种轻量的方案就不太适用了。
+
+`agent` 和 `send` 的模型更关注**数据处理**，毕竟 `agent` 在创建时就需要指定初始值，而 `send` 的作用是给 `agent` 发送一个操作，对其中的数据做操作。再者，`agent` 天然就是并发安全的，可以减少自己加锁的麻烦。处理数量可控且相对轻量的数据时，就可以用 `pmap`。
+
+`go` 和 `chan` 就跟它们在 Go 语言里的地位一样，如果要开发**高并发**的复杂系统，或者管理系统的复杂性，将组件**解耦**，就可以抱着抛弃函数式风格的代价使用。`core.async` 其实遭到了[批评](https://stackoverflow.com/questions/18198666/isnt-core-async-contrary-to-clojure-principles)，而且的确有一些问题。Stack Overflow 上的 [aeuhuea](https://stackoverflow.com/users/1846370/aeuhuea) 就举了个很典型的例子：
+
+```clojure 
+(defn take-and-print [c]
+ (println (<! c)))
+
+(def ch (chan 1))
+(>!! ch 123)
+
+(go (take-and-print ch))
+;; Assert failed: <! used not in (go ...) block
+```
+
+上面的代码报错了，因为 `<!` 和 `>!` 如果被放在函数里面，就会被认定没有按规定放在 `go` block 里，哪怕函数就是在 `go` 里面调用的。我一般的解决方法是把 `<!` 取出的值作为参数传递给函数，或者用宏。
+
+## 最后
+
+异步编程很难不考虑状态，我还真的有点好奇 Haskell 作为纯函数式编程语言，不像 Clojure 这样容忍一部分的可变状态和副作用，是怎么实现异步的。我得找时间去研究研究。
+
+这些概念可能不太好理解，我希望我讲清楚了。如果你有任何问题，或者发现了我有疏漏的地方，欢迎给我发邮件。那么，回见！
+
+[^1]: 其实可以给 `agent` 发这样的函数：`(fn [_] "new value")`
+
+[^2]: 说实话，很少用 `go` 返回的 *channel* 取返回值，因为不如 `future` 和 `deref` 方便。
+
+[^3]: 在 Go 语言里，空结构体 `struct{}` 不占空间，所以经常用来做信号量放进 `channel` 里。Clojure 的空映射 `{}` 底层是个 Java 类，不可能不占空间，这里只是我自己的习惯，你可以换成 `true`、`:signal` 之类的，但是不能放 `nil`，会报错，因为 `nil` 在某些操作里表示 *channel* 关闭了或者空了。
+
+[^4]: 即协程，是 Go 语言建立在线程之上的用户级、轻量的并发单元，用 `go` 启动。Clojure 里的 `go` 仅仅是借其形式，不见得和 *goroutine* 一样轻量（我们可是在 Java 虚拟机上啊！）。
+
+[^5]: 这个例子由 Mark McGranaghan 和 Eli Bendersky 编写，用 [CC BY 3.0 协议](https://creativecommons.org/licenses/by/3.0/)授权。我只截取了其中的一部分。
 
-[^1]: 不过其实也可以给 `agent` 发这样的函数：`(fn [_] "new value")`
+[^6]: 唯一的缺陷是，如果程序意外退出，而 *channel* 里还有些日志没写入数据库，这些数据就丢失了。