clojure - Core.async:从 promise-chans 集合中获取所有值

Question

考虑这样一个数据集:

(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 可能是一个很好的工具。可以为每个项目设置单独的 channel ，也可以使用一个 channel 。我不确定这里哪种方式更可取。对于多个 channel ，可以使用传感器进行后处理/解析。还有一个特殊的 promise-chan 在这里可能是合适的。

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

(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))))

Answer 1

我想说 promise-chan 在这里弊大于利。问题在于，大多数 core.async API(a/merge、a/reduce 等)都依赖于 channel 将在以下时间关闭的事实某些时候，promise-chan 又永远不会关闭。

因此，如果坚持使用 core.async 对您来说至关重要，那么更好的解决方案是不使用 promise-chan，而是使用普通 channel ，这将是在第一次 put! 之后关闭:

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

此时，您使用的是封闭 channel ，事情变得简单了一些。要收集所有值，您可以这样做:

;; (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 channel 作为 promise (以 promise-chan 的形式或在单个 put 后​​关闭的 channel ! ) 不是最好的方法。当事情发展时，事实证明 core.async API 总体上(您可能已经注意到)并不像它应该的那样令人愉快。还有几个unsupported constructs ，这可能会迫使您编写比实际更少的惯用代码。此外，没有内置的错误处理(如果在 go-block 中发生错误，go-block 将静默返回 nil)并且要解决这个问题，您需要自己想出一些办法(重新发明轮子)。因此，如果您需要 promise ，我建议为此使用特定的库，例如 manifold或 promesa .