如何清理 clojure core.async 通道？

Question

我是第一次看 Clojure core.async，正在阅读 Rich Hickey 的精彩演示：http://www.infoq.com/presentations/clojure-core-async

我对他在演示结束时展示的示例有疑问：

根据 Rich 的说法，这个示例主要是尝试获取特定查询的网络、视频和图像结果。它为每个结果并行尝试两个不同的来源，并为每个结果提取最快的结果。整个操作可能需要不超过 80 毫秒，所以如果我们不能得到例如80ms的图像结果，我们就放弃了。 'fastest' 函数创建并返回一个新通道，并启动两个 go 进程竞相检索结果并将其放在通道上。然后我们从“最快”通道中取出第一个结果并将其放到 c 通道上。

我的问题：在我们获得第一个结果后，这三个临时的、未命名的“最快”频道会发生什么？据推测，仍然有一个 go 进程试图将第二个结果放到通道上，但没有人在听，所以它实际上从未完成。而且由于通道永远不会绑定到任何东西，所以我们似乎再也没有办法用它做任何事情了。 go 进程和通道会“意识到”没有人再关心他们的结果并清理自己吗？还是我们本质上只是在这段代码中“泄露”了三个通道/go 进程？

Answer 1

没有泄漏。

停放的gos 被附加到他们试图执行操作的通道上，除此之外没有独立存在。如果其他代码对某个go 停驻的频道失去兴趣（注意，如果go 停在alt! / alts! 上，则go 可以同时成为多个频道的推杆/接球手），那么最终它'将与这些频道一起被 GC。

唯一需要注意的是，为了获得 GC，gos 实际上必须先停车。所以任何go 一直在循环中做事而没有停车（<! / >! / alt! / alts!）实际上将永远存在。不过，这样的代码很难偶然写出。

Answer 2

除了注意事项和例外，您可以在 REPL 上测试 JVM 上的垃圾收集。

例如：

(require '[clojure.core.async :as async])
=> nil

(def c (async/chan))
=> #'user/c
(def d (async/go-loop [] 
         (when-let [v (async/<! c)] 
           (println v) 
           (recur))))
=> #'user/d

(async/>!! c :hi)
=> true
:hi        ; core.async go block is working

(import java.lang.ref.WeakReference)
=> java.lang.ref.WeakReference    ; hold a reference without preventing garbage collection
(def e (WeakReference. c))
=> #'user/e
(def f (WeakReference. d))
=> #'user/f

(.get e)
=> #object[...]
(.get f)
=> #object[...]

(def c nil)
=> #'user/c
(def d nil)
=> #'user/d
(println "We need to clear *1, *2 and *3 in the REPL.")
We need to clear *1, *2 and *3 in the REPL.
=> nil
(println *1 *2 *3)
nil #'user/d #'user/c
=> nil
(System/gc)
=> nil
(.get e)
=> nil
(.get f)
=> nil

刚刚发生了什么？我设置了一个 go 块并检查它是否正常工作。然后用WeakReference观察通信通道（c）和go块返回通道（d）。然后我删除了对 c 和 d 的所有引用（包括我的 REPL 创建的 *1、*2 和 *3），请求垃圾收集，（幸运的是，System.gc Javadoc 没有做出强有力的保证）然后观察我的弱引用已被清除。

至少在这种情况下，一旦删除了对所涉及通道的引用，这些通道就会被垃圾收集（不管我是否未能关闭它们！）

Answer 3

假设fastest产生的通道只返回最快查询方法的结果，然后关闭。

如果产生第二个结果，您的假设可能是fastest 进程已泄漏。他们的结果永远不会被消耗。如果他们依赖于所有结果被消耗来终止，他们就不会终止。

请注意，如果在alt! 子句中选择了频道t，也会发生这种情况。

解决此问题的通常方法是使用close! 关闭最后一个go 块中的通道c。然后，对封闭通道进行的看跌将被丢弃，并且生产者可以终止。

这个问题也可以通过fastest的实现来解决。在fastest 中创建的进程本身可以通过alts! 和timeout 进行put，如果在一定时间内没有消耗产生的值，则终止。

我猜 Rich 没有解决幻灯片中的问题，而是使用了一个不太长的示例。