等待 core.async 的 n 个频道

Question

以同样的方式alt!等待其中一个 n个通道获取值，我正在寻找等待[=16=的惯用方式]alln个通道取值

我需要这个，因为我 "spawn" n go blocks 来处理异步任务，我想知道它们什么时候完成。我相信有一种非常漂亮的方法可以实现这一点。

Answer 1

你可以说 (mapv #(async/<!! %) channels).

如果您想在单个值到达时对其进行处理，然后在最终通道产生一个值后做一些特殊的事情，您可以利用 alts! / alts!! 采用向量这一事实的通道，它们是函数，而不是宏，所以你可以很容易地传入动态构造的向量。

因此，您可以使用 alts!! 等待您最初收集的 n 个频道，然后在剩余的频道等上再次使用它。

(def c1 (async/chan))
(def c2 (async/chan))

(def out
  (async/thread
    (loop [cs [c1 c2] vs []]
      (let [[v p] (async/alts!! cs)
            cs (filterv #(not= p %) cs)
            vs (conj vs v)]
        (if (seq cs)
          (recur cs vs)
          vs)))))

(async/>!! c1 :foo)
(async/>!! c2 :bar)

(async/<!! out)
;= [:foo :bar]

如果您想从所有输入通道获取所有值，然后在它们全部关闭时执行其他操作，您需要使用 async/merge:

clojure.core.async/merge
([chs] [chs buf-or-n])
Takes a collection of source channels and returns a channel which contains all values taken from them. The returned channel will be unbuffered by default, or a buf-or-n can be supplied. The channel will close after all the source channels have closed.

Answer 2

使用core.asyncmap函数：

(<!! (a/map vector [ch1 ch2 ch3]))
;; [val-from-ch-1 val-from-ch2 val-from-ch3]

等待 core.async 的 n 个频道

Waiting for n channels with core.async

clojure

clojurescript

core.async