clojure - core.async 中消息在订阅之前发布

Question

在下面的示例中，我可以看到，已发布的消息到达了订阅的 channel ，尽管它们是在订阅之前发布的。

(let [in (async/chan)
      out (async/chan)
      pub (async/pub in :key)]

  (async/go
    (>! in {:id 1 :key :k1})  
    (>! in {:id 2 :key :k1})  
    (>! in {:id 3 :key :k1}))

    (async/sub pub :k1 out)

    (async/go-loop []
      (println (<! out)) 
      (recur)))

这是预期的行为吗？据我在文档中看到，它明确指出:

Items received when there are no matching subs get dropped.

我在 Clojure 和 ClojureScript 中得到相同的结果。

添加: 使用 mult/tap 我看到类似的行为

Answer 1

您不知道消息是在订阅之前发布的。因为 go 是异步的，所以订阅很可能在第一条消息放入 channel 之前发生。这确实是一种竞争条件。

尝试在订阅之前添加一个(Thread/sleep [some value])，看看会发生什么。