使用core.async等待n个通道

Question

以同样的方式alt!等待的一个 n个信道得到的值,我正在寻找的习惯的方法来等待所有的 N个信道得到的值.

我需要这个,因为我"生成"n块来处理异步任务,我想知道它们何时完成.我确信有一种非常漂亮的方法来实现这一目标.

Answer 1

使用core.async map功能:

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

Answer 2

你可以说(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])
获取源通道的集合并返回包含从它们获取的所有值的通道.默认情况下,返回的通道将是无缓冲的,或者可以提供buf-or-n.所有源通道关闭后,通道将关闭.