如何使用 core.async 代替回调？答案

【问题标题】：How to use core.async in place of callbacks?如何使用 core.async 代替回调？
【发布时间】：2018-12-30 11:00:26
【问题描述】：

我在 ClojureScript 上使用 core.async 以避免使用 node.js 回调。问题是我达到了 1024 条待处理消息的限制。

为避免这种情况，我需要将所有消息发送到同一 go 块内的通道。但这在 core.async 上是不可能的，因为匿名函数会使go 的效果无效，所以我不能这样做：

(go
  (. socket on "data" #(>! chan %)))

那么，有没有办法绕过这个限制？

【问题讨论】：

我刚刚添加了适当的语言提示。使用 clojurescript 和 node.js 或 javascript 进行标记时，请使用提示以确保语法高亮显示正常工作。

标签： node.js callback clojurescript core.async

【解决方案1】：

我不确定我是否理解为什么它必须在同一个 go 块中。通常，你会这样做：

(. socket on "data" #(go (put! chan %)))

然后在另一个 go 块中处理 chan

【讨论】：

当背压太大时失败，错误AssertionError Assert failed: No more than 1024 pending puts are allowed
事实上，上面的代码不会抛出异常，因为它在go 块内。但是，如果您将 put! 包裹在 (try... (catch Throwable t ...)) 周围，您会看到错误

【解决方案2】：

为了模拟错误，我们可以“模拟”一种回调：

(let [callback (atom nil)
      on-data (fn [fun]
                (reset! callback fun))
      chan (async/chan)]

如果我们尝试使用(on-data #(async/go (async/put! chan %))) 添加回调，则会超出限制。另外，由于我们使用async/go，它会导致频道内的消息乱序。

我发现解决这个问题的唯一方法是在atom 内创建一个无限的promise-chan 列表，每个回调都会获取第一个元素，发布一条消息，然后删除列表中的第一个元素。然后，我们可以在 go 块中添加一个 doseq，它会为我们发布消息：

(let [inf-list (atom (map (fn [_] (async/promise-chan)) (range)))]
  ; We need to iterate over inf-list before anything
  (let [lst @inf-list]
    (async/go
     (doseq [c lst]
       (async/>! chan (async/<! c)))))

  (on-data #(do
              (async/put! (first @inf-list) %)
              (swap! inf-list rest))))

【讨论】：