Core.async：从 promise-chans 集合中获取所有值答案

【问题标题】：Core.async: Take all values from collection of promise-chansCore.async：从 promise-chans 集合中获取所有值
【发布时间】：2016-09-17 23:08:44
【问题描述】：

考虑这样的数据集：

(def data [{:url "http://www.url1.com" :type :a}
           {:url "http://www.url2.com" :type :a}
           {:url "http://www.url3.com" :type :a}
           {:url "http://www.url4.com" :type :b}])

应该并行请求这些 URL 的内容。根据项目的 :type 值，这些内容应由相应的函数解析。一旦所有响应到达，解析函数就会返回集合，这些集合应该被连接起来。

所以我们假设有函数parse-a 和parse-b，当它们被传递一个包含HTML 内容的字符串时，它们都返回一个字符串集合。

看起来core.async 可能是一个很好的工具。一个人可以为每个项目或一个单独的通道具有单独的通道。我不确定在这里哪种方式更可取。对于多个通道，可以使用传感器进行后处理/解析。还有一个特殊的promise-chan 可能在这里合适。

这是一个代码草图，我正在使用基于回调的HTTP kit 函数。不幸的是，我在 go 块中找不到通用解决方案。

(defn f [data] 
  (let [chans (map (fn [{:keys [url type]}] 
                     (let [c (promise-chan (map ({:a parse-a :b parse-b} type)))] 
                       (http/get url {} #(put! c %))
                       c))
                   data)
        result-c (promise-chan)] 
    (go (put! result-c (concat (<! (nth chans 0))
                               (<! (nth chans 1))
                               (<! (nth chans 2))
                               (<! (nth chans 3)))))
    result-c))

结果可以这样读取：

(go (prn (<! (f data))))

【问题讨论】：

您的问题具体是什么？

标签： clojure core.async

【解决方案1】：

我想说promise-chan 在这里弊大于利。问题是大多数core.async API（a/merge、a/reduce 等）都依赖于通道会在某个时候关闭的事实，而promise-chans 则永远不会关闭。

所以，如果坚持core.async对你来说很重要，更好的解决方案是不要使用promise-chan，而是使用普通频道，在第一个put!之后关闭：

...
(let [c (chan 1 (map ({:a parse-a :b parse-b} type)))]
  (http/get url {} #(do (put! c %) (close! c)))
  c)
...

此时，您正在使用封闭的渠道，事情变得有点简单。要收集所有值，您可以执行以下操作：

;; (go (put! result-c (concat (<! (nth chans 0))
;;                            (<! (nth chans 1))
;;                            (<! (nth chans 2))
;;                            (<! (nth chans 3)))))
;; instead of above, now you can do this:
(->> chans
     async/merge
     (async/reduce into []))

UPD（以下是我的个人意见）：

似乎，使用core.async 频道作为承诺（以promise-chan 的形式或在单个put! 之后关闭的频道）不是最好的方法。随着事情的发展，事实证明core.async API 总体上（您可能已经注意到）并没有想象中那么令人愉快。还有几个unsupported constructs，这可能会迫使您编写的惯用代码比可能的要少。此外，没有内置的错误处理（如果在go-block 中发生错误，go-block 将静默返回nil），要解决这个问题，您需要自己想办法（重新发明轮子）。因此，如果您需要 Promise，我建议您为此使用特定的库，例如 manifold 或 promesa。

【讨论】：

实际上，与此同时，我或多或少地想到了您提出的解决方案。在我决定为其添加错误处理之前，这很顺利。我可能会改用简单的承诺。

【解决方案2】：

我也想要这个功能，因为我真的很喜欢core.async，但我也想在某些地方使用它，比如传统的JavaScript promises。我想出了一个使用宏的解决方案。在下面的代码中，<? 与<! 相同，但如果出现错误则会抛出。它的行为类似于Promise.all()，如果它们都成功，它会返回来自通道的所有返回值的vector；否则它将返回第一个错误（因为<? 将导致它抛出该值）。

(defmacro <<? [chans]
  `(let [res# (atom [])]
     (doseq [c# ~chans]
       (swap! res# conj (serverless.core.async/<? c#)))
     @res#))

如果您想查看函数的完整上下文，它位于GitHub。它的灵感来自 David Nolen 的 blog post。

【讨论】：

【解决方案3】：

在async.core 中使用pipeline-async 来启动异步操作，例如http/get，同时以与输入相同的顺序传递结果：

(let [result (chan)] 
  (pipeline-async 
    20 result
    (fn [{:keys [url type]} ch]
      (let [parse ({:a parse-a :b parse-b} type)
            callback #(put! ch (parse %)(partial close! ch))]  
        (http/get url {} callback)))
    (to-chan data))
  result)

【讨论】：

【解决方案4】：

如果有人还在看这个，请补充@OlegTheCat 的答案：

您可以使用单独的通道来处理错误。

(:require [cljs.core.async :as async]
            [cljs-http.client :as http])
(:require-macros [cljs.core.async.macros :refer [go]])

(go (as-> [(http/post <url1> <params1>)
           (http/post <url2> <params2>)
           ...]
          chans
          (async/merge chans (count chans))
          (async/reduce conj [] chans)
          (async/<! chans)
          (<callback> chans)))

【讨论】：