Coffee Break Clojure, Async

上一篇 我提及了如何用 Ring 开发 HTTP 服务器，也提到 Ring 提供了同步 API 和异步 API。那么，什么是异步编程？为什么需要异步？在什么情况下需要让一个程序同时做很多事情？

同步、异步、并发和并行

同步（synchronization）的中文名字可能会让人误以为是多个实体同时进行操作，实际上同步的意思是指系统中所有事件都依照相同的顺序进行协调。可以用工厂流水线举例，假如流水线中一个工人需要等上一个工人交付一部分工作之后才能干活，整个流水线都要所有人一起干活才能转动起来，这个流水线就是同步的。反之，假如某些工作可以分开做，比如一部分人负责生产 A 零件，另一部分人负责生产 B 零件，两者互不干扰，而最终把 A 和 B 组装在一起的又是另一拨人，这个系统就是异步的（asynchronous）。

异步（asynchrony）和并发（concurrency）很容易搞混，并不是同一个概念。异步是指系统中各个事件的状态不同步，而并发通常是指多个线程（或类似线程的东西）同时做类似的操作。

以下分别是异步和并发的例子：

1. 小明和小华准备去桌游店玩，小明负责去买饮料和零食，而小华先去桌游店占位置。他们的状态是不同步的，小华并不需要等到小明买完东西才能去桌游店。不过，小明和小华最终会在桌游店汇合，他们一起玩桌游时，就又回到了同步状态。 异步通常是程序里的一小部分，如果系统中所有任务都是异步的，那这些任务大概就不应该放在同一个系统里了，因为状态之间没有同步的必要。
2. 某推销团队在城市某片区招募了 100 名推销员，给他们派发相同的任务：向路人介绍产品，并尽可能获客。从整个推销计划的角度来看，这是个并发度为 100 的系统，增加并发度能够提高单位时间内的销售成绩，减少并发度效果则相反。也可以说推销员之间是异步的，但系统整体是同步的，销售成绩最后都会汇总到一个地方。

简单来说，异步是局部状态，对系统整体影响不大，而并发则事关整个系统的执行效率，是更宏观的概念。

从编程实践来看，并发往往是更大更复杂的系统才考虑的（比方说网络爬虫必然不能一次只爬一个网页），而异步则是较小的程序会采用的轻量方案（比方说网页渲染中加载某些资源时为了不阻碍主线程，所以要异步加载）。

至于并行（parallelism），这个概念经常在操作系统的语境里和并发做区分。并行是指程序同时运行在多个 CPU 核心上，是物理意义上的同时发生；并发是指 CPU 调度器时进程交替执行，由于交替速度非常快，所以即便是在单核机器上，也能造成程序在同时运行的假象。

脱离操作系统的语境，这种区分的意义就不大了。在编程的语境下，并行是指程序在同时进行的事实（开发者很少在乎程序究竟是在多核还是在单核机器上运行的，这对编写程序而言不重要），而并发则是组织程序的一种方式。

接下来就看看如何在 Clojure 里组织异步和并发。

面向未来编程

前文提到，异步是局部状态，系统最终还是要同步信息的。许多编程语言里的做法是，在当前线程（thread）里启动子线程，相关的线程都持有对同一块内存区域的引用，他们在这一块内存区域里读写数据。之前我提到，Clojure 是 函数式编程语言 ，不允许共享状态，那怎么办呢？

假设我们需要从某个 API 获取数据，但问题是网络状态不稳定，这个操作可能很费时间。与此同时，我们还要处理手头已有的数据。

;; 这里用到了 https://github.com/dakrone/clj-http
(require '[clj-http.client :as client])
(let [response (client/get "https://example.com")
 result1 (process-data (:body response))
 result2 (process-data some-data-we-already-have)]
 (merge result1 result2)))

上面这段代码会在求值 client/get 这个形式的时候被阻塞，直到收到 HTTP 响应体。假设 process-data 操作也要费时间，这个程序的效率就很低，因为在等待网络传输的时间里，程序本身什么都没做。

如果不能提前定义 result1 变量，然后再异步执行完 client/get 之后写入 result1 变量，主线程怎么拿到数据呢？

这时可以用到 future 和 deref。future 宏接收一系列表达式，然后立刻返回一个 ref（引用），在之后，我们可以用 deref 尝试获取 ref 指向的值。如果 ref 指向的值还不能获取，就阻塞调用 deref 的线程。

(require '[clj-http.client :as client])
(let [response (future (client/get "https://example.com"))
 result2 (process-data some-data-we-already-have)
 result1 (process-data (:body (deref response)))]
 (merge result1 result2)))

把 client/get 包裹在 future 宏里之后，我们立刻得到了 response，这个 response 并不是实际的 HTTP 响应，而是对未来能够获取到的数据的引用。

拿到这个引用之后，我们立刻处理手头已有的数据。等这些数据处理完成之后，尝试 deref（解引用）一开始拿到的引用，此时 HTTP 请求已经发出有一段时间了，我们有可能直接就能拿到数据，然后就可以开始处理了；如果没拿到，也只需要再等一会儿，比一开始就等着什么也不做强多了。

持有对未来可用数据的引用，和持有对某一块内存区域的“引用”，有什么区别呢？

可以发现上述程序没有状态变更，future 返回了一个值，deref 不过以 future 的返回值作为输入，返回了新的值，数据流很清晰，很函数式。此外，它还很优雅地把程序从异步状态恢复到了同步状态，不需要我们自己写同步逻辑。

以这个 Go 程序为反例：

package main

import (
	"fmt"
	"time"
)

func main() {
 // 定义一个变量用来存放数据
 result := ""
 // 异步执行函数，两秒后给 result 赋值为 "OK"
 go (func() {
 time.Sleep(2 * time.Second)
 result = "OK"
 })()
 // 轮询，直到 result 不是空字符串才退出循环
 for {
 if result != "" {
 fmt.Println(result)
 break
 }
 }
}

这段程序既要管理状态，又要轮询状态，非常难读。

Go 语言也有自己的异步编程范式，实际上不必这么写。之后我们就会看到 Go 的异步编程，因为 Clojure 也借鉴了 Go 语言。

其实我本来想用 JavaScript 写，但我在浏览器里执行的时候，我的页面竟然卡死了。后来我才知道浏览器的 JS 运行时是单线程的，用于轮询的无限循环把主线程卡死了，JS 根本没时间管子线程，于是赋值语句就一直没执行。JavaScript 也有类似于 future 的 async 关键词，async 函数会返回 Promise，类似于 ref，而 deref 对应的是 await，详见 MDN: async function 。

如果你熟悉 JavaScript 的 async 和 await，那 Clojure 的 future 和 deref 就很容易理解。

此外，(deref ref) 可以简写为 @ref。不过用 @ 语法糖的话就用不了 deref 的超时控制。(deref ref) 默认会一直等待，如果 ref 的值一直没来，程序就会卡死；可以用 (deref ref timeout fallback) 做超时控制，如果经过了 timeout 毫秒还没能解引用成功，就用 fallback 值。

在继续之前我还要提醒对异步编程不那么熟悉的读者，无论是哪种异步机制，无论怎么实现并发，线程的执行顺序都是不固定的。

(dotimes [number 10] (future (println number)))
;; 05
;; 423
;;
;; 8
;; 9
;; nil1
;; 7
;;
;;
;; 6

上面的输出很乱，原因在于 println 的换行符何时打印也不能确定，dotimes 本身的返回值 nil 也混在这些换行得乱七八糟的数字当中。

有办法能控制并发的执行顺序，但要保证顺序必然要有等待、排队的过程，速度就有损失，甚至有可能变成串行的。如果要严格同步，那就不应该异步——是的，这是句废话。总之，需要严格保证执行顺序的操作不该异步。

不阻塞的异步

尝试解引用 future 返回的引用可能会阻塞当前线程，如果希望当前线程不被阻塞呢？有时候我们并不需要严格同步数据，如果新值没有到达，那就用旧的值。

比方说我们有个计数器，每 5 秒给计数器增加 1，解引用这个计数器就拿到当前的数，而不是等五秒过后再拿新的数。

要这么做，可以用 agent：

(def counter (agent 0))
;; => #'user/counter

@counter
;; => 0 

(future 
 (doseq [_ (range)] 
 (Thread/sleep 5000) 
 (send counter inc)))
;; => #object[clojure.core$future_call$reify__8574 0x4abf3f0 {:status :pending, :val nil}]

@counter
;; => 0 

;; 五秒过后 
@counter
;; => 1

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 用于异步编程，我们完全可以违反「不允许可变状态」的规则，做这种事情：

(def counter (agent 0))
;; => #'user/counter

(deref counter)
;; => 0 

(send counter inc)
;; => #object[clojure.lang.Agent 0x2f6bbeb0 {:status :ready, :val 1}]

(deref counter)
;; => 1

(send counter inc)
;; => #object[clojure.lang.Agent 0x2f6bbeb0 {:status :ready, :val 1}]

(deref counter)
;; => 2

这就要靠程序员的自觉了。不过，agent 其实没那么好用，因为只能发送函数，而不能替换值本身，用途很有限。

agent 可以存任何类型的值，也可以存集合类型，比方说我们可以一次性给 agent 发多个函数，而每个函数的返回值都是由旧集合拼接成的新集合，就可以并发地处理多个数据。比方说，下面是一个简单的并发爬虫，把数个 HTML 页面塞进了同一个向量里：

(defn crawl [url] 
 (:body (client/get url)))

(defn crawl-append [old url] 
 (conj old (crawl url)))

(def crawler (agent []))

(def urls ["https://chlo.is"
 "https://www.geedea.pro"
 "https://blog.cytrogen.icu/"])

(doseq [url urls]
 ;; crawler agent 里面已经有的值会作为 crawl-append 的第一个参数 
 ;; 后续传入的可变数量的参数（如这里的 url），会变成 crawl-append 剩下的参数
 (send crawler crawl-append url))

@crawler
;; 会返回由当前已经爬取完成的 HTML 字符串组成的向量

对于这种结构简单、数量确定的操作，还可以使用 pmap（parallel map），也就是并行版本的 map。下面这段代码是并行的（准确来说是并发），不需要 agent 也不需要 deref。

(pmap crawl urls)
;; 会返回包含所有爬取结果的惰性序列

pmap 返回的是“半”惰性序列，这些 crawl 其实已经执行了，只不过没有把值返回过来，只有在读取惰性序列的时候才会得到值。pmap 底下用的是 future。

如果要用 map 处理耗时较长的任务，就可以考虑换成 pmap 同时处理。

;; A function that simulates a long-running process by calling Thread/sleep:
(defn long-running-job [n]
 (Thread/sleep 3000) ; wait for 3 seconds
 (+ n 10))

;; Use `doall` to eagerly evaluate `map`, which evaluates lazily by default.

;; With `map`, the total elapsed time is just under 4 * 3 seconds:
user=> (time (doall (map long-running-job (range 4))))
"Elapsed time: 11999.235098 msecs"
(10 11 12 13)

;; With `pmap`, the total elapsed time is just over 3 seconds:
user=> (time (doall (pmap long-running-job (range 4))))
"Elapsed time: 3200.001117 msecs"
(10 11 12 13)

上面这个例子来自 ClojureDocs 。

你可能好奇，如果 pmap 处理大量的数据，比方说一不小心传了个无穷序列进去怎么办？这会同时处理无穷个数据，把 CPU 撑爆吗？老实说，我不知道具体会发生什么，但大概率 Clojure 或 Java 会控制最大并发数，根据当前负载，把过多的请求串行化以减轻压力。

无论有没有这种说法，程序自己完全不控制并发数量也是不行的。无论是 future 还是 agent，貌似都没有提供很好的并发控制机制，难道只能一开始就少派发任务吗？如果不确定有多少任务要运行呢？

Gopher？是你吗？

命名空间 clojure.core.async 提供了另外一种异步编程的方式，但没有放在标准库里，必须要添加依赖才能使用。

;; deps.edn 
org.clojure/core.async {:mvn/version "1.9.865"}
;; leiningen
[org.clojure/core.async "1.9.865"]

core.async 几乎和 Go 语言异步编程的要素一模一样，尤其是 go block——把语句放在 go block 里面，语句就会异步执行。把消息队列 chan 当作一等公民的做法，也如出一辙。根据 Clojure 官方的博客文章 ，这种异步编程风格其实有一段历史了，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.

无论如何，让我们来看看怎么用 go 和 chan 实现异步编程吧。如果你已经会写 Go，那么接下来的代码会很熟悉。

(require '[clojure.core.async :refer [go chan >! <!]])

(def queue (chan))

(go 
 (>! queue "Hello?"))

(go 
 (println (<! queue)))

上述例子中，我们启动了两个 go 线程，一个线程把 "Hello" 放进 queue 里，另一个线程把 "Hello" 取出来。

chan 函数用于创建 channel（频道），可以理解为消息队列。对于 channel，有放（put）和拿（take）的操作，分别用 >!（放）和 <!（拿）表示。这里的 ! 类似于 ?，是 Clojure 的命名习惯，用于标识会对状态进行变更的操作。说到底，我们的确对状态进行了变更，上述语句缺乏引用透明性，不够函数式。

chan 可以设置缓冲区大小，(chan 5) 就是缓冲区大小为 5 的队列，一旦队列满了，再试图往里面放消息，放消息的线程就会被阻塞。

(def only-put-once (chan 1))

(go (>! only-put-once 1)
 ;; 下面这个操作会阻塞这个 go 线程
 (>! only-put-once 1))

反之，如果队列空了还试图从里面拿东西，就会阻塞拿消息的线程。

(def empty-chan (chan))

(go (<! empty-chan)
 ;; 除非有人往 empty-chan 里放东西
 ;; 否则下面这句话不会被打印
 (println "Help! I'm blocked!"))

go 在 Clojure 里是个宏，在 Go 里是个放在函数调用之前的关键字，效果都是让语句异步执行。不过 Clojure 中的 go 会返回一个 channel，go block 里的语句执行完成后，会把返回值通过 channel 传递回来。

放和拿的操作可以阻塞线程，因此可以用作并发控制。

学过操作系统的读者可能知道信号量（semaphore），我们可以用 chan 实现简单的信号量。比方说，我们先给一个 chan 存入 50 个信号，然后启动不定数量的 go 线程，但在启动线程之前，派发器必须要拿到信号量才行。如果信号量不够（尝试从空队列里取数据），就会阻塞派发器所在的线程，直到有信号量被回收。

(require '[clojure.core.async :refer [go chan put! <!! >!]])
;; 定义信号量
(def semaphore (chan))
;; 将 50 个空 map 放入信号量中
(dotimes [_ 50]
 (put! semaphore {}))
;; 启动一个无限循环派发器
(doseq [number (range)]
 ;; 尝试获取信号量，如果已经消耗完 50 个，阻塞派发器
 (<!! semaphore)
 ;; 获取到信号量时，启动 go 线程
 (go (Thread/sleep (rand-int 10000))
 (println number)
 ;; 线程工作完成后，把一个信号放入 semaphore
 (>! semaphore {})))

上述程序永远不会同时运行超过 50 个 go 线程。也可以反着实现，定义大小为 50 的 channel，每派发一个线程之前都要放个东西到 channel 里，比如 {}，放满了话的话就阻塞。不妨试着在 REPL 里写一写试试？

需要注意的是放和拿的操作有很多变体，在不同场景下使用。之前提到的 >! 和 <! 只能放在 go block 中，放在其他地方会报错。

在 go block 外，可以用 >!! 和 <!!，一般不建议用，因为会阻塞主线程。更好的选择是 put!（放）和 take!（拿），这两个操作是异步的，无论放没放进去、拿没拿出来，都不会阻塞当前的线程（但是会阻塞异步的子线程）。

由于 take! 是异步的，不会立刻返回值，所以得传一个回调函数给它，这样在能拿到值的时候，就会调用回调函数。下面是 ClojureDocs 的示例：

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 和 channel 实现各种并发处理管线。我刚才已经演示了两个线程（协程）用 channel 异步通信，演示了派发器和并发控制。我们还可以把更多 Go 语言的编程模式搬到 Clojure 里来。

我前面提到，core.async 提供的这套异步编程模式很不函数式，因为 channel 就是存放状态的队列，与之相关的放和拿的操作也不具有纯函数应有的引用透明性。

不过这是否值得呢？接下来我要用 Go By Example 网站上的 Stateful Goroutines 的例子回答这个问题。

 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，稍后把处理结果发送回去。

 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）。

;; (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 之后就可以继续做自己的事情，不必等着把日志写进数据库。

把副作用、状态管理和 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 其实遭到了 批评 ，而且的确有一些问题。Stack Overflow 上的 aeuhuea 就举了个很典型的例子：

(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 协议 授权。我只截取了其中的一部分。
6. 唯一的缺陷是，如果程序意外退出，而 channel 里还有些日志没写入数据库，这些数据就丢失了。