什么是替代的最佳方法！在通道向量上多次？

Question

我正在使用 core.async 并行执行某些操作，然后使用 alts!! 等待一定数量的超时结果。

(ns c
  (:require [clojure.core.async :as a]))

(defn async-call-on-vector [v]
  (mapv (fn [n]
          (a/go (a/<! (a/timeout n)) ; simulate long time work
                n))
        v))

(defn wait-result-with-timeout [chans num-to-get timeout]
  (let [chans-count (count chans)
        num-to-get (min num-to-get
                        chans-count)]
    (if (empty? chans)
      []
      (let [timeout (a/timeout timeout)]
        (loop [result []
               met 0]
          (if (or (= (count result) num-to-get)
                  (= met chans-count)) ; all chan has been consumed
            result
            (let [[v c] (a/alts!! (conj chans timeout))]
              (if (= c timeout)
                result
                (case v
                  nil (do (println "got nil") (recur result met)) ; close! on that channel
                  (recur (conj result v) (inc met)))))))))))

然后像这样调用：

user=> (-> [1 200 300 400 500] c/async-call-on-vector (c/wait-result-with-timeout 2 30))

这个表达式会打印出很多got nil。似乎 go 块返回的通道将在返回结果后关闭该通道。这将导致alts!! 在这种情况下返回 nil。但这对 CPU 很不友好，就像忙着等待一样。有没有办法避免这种情况？

我通过定义一个像 go 这样的宏来解决这个问题，但是返回一个不会在返回结果时关闭的通道。这是解决问题的正确方法吗？

Answer 1

我正在使用 core.async 并行执行某些操作，然后使用 alt！！等待一定数量的结果超时。

您似乎想要收集将由某些通道传递的所有值，直到所有这些通道都关闭或发生超时。一种方法是将merge 这些通道放到一个通道上，然后在go-loop 中使用alts! 将这些值收集到一个向量中：

(defn wait-result-with-timeout [chans timeout]
  (let [all-chans (a/merge chans)
        t-out (a/timeout timeout)]
    (a/go-loop [vs []]
      (let [[v _] (a/alts! [all-chans t-out])]
        ;; v will be nil if either every channel in
        ;; `chans` is closed, or if `t-out` fires.
        (if (nil? v)
          vs
          (recur (conj vs v)))))))

go 块返回的通道似乎会在返回结果后关闭该通道。

你是对的，这是一个 go 块的记录行为。

我通过定义一个像 go 这样的宏来解决这个问题，但是返回一个不会在返回结果时关闭的通道。这是解决问题的正确方法吗？

可能不会，尽管我不能说它对您的特定用例是对还是错。一般来说，如果通道完成交付值，则应该关闭，以指示完成交付值的语义。例如，上面的代码使用all-chans的关闭来表示没有更多的工作等待。