如何耗尽通道的值然后返回结果（ClojureScript）？

Question

假设频道chan 在队列中具有值“1”和“2”。

目标： 创建一个函数，该函数接受chan 并返回向量[1 2]。请注意，如果此函数必须在返回值之前阻塞一段时间，我完全没问题。

尝试：

(defn chan->vector
  [chan]  
  (let [a (atom true) v []]
    (while (not-nil? @a)
      (go
        (reset! a (<! chan))
        (into v @a)
        (reset! a (<! chan))
      )
    ) v
  )
)

结果：我的 REPL 冻结并最终吐出一个巨大的错误。我开始意识到这是因为 (go ...) 块是异步的，因此会立即返回。因此，我的(while ...) 循环中的原子永远不会被设置为nil，并且循环永远不会终止。

那么我怎样才能达到预期的效果呢？如果它是相关的，我正在使用 ClojureScript 并以 nodejs 为目标。

Answer 1

您应该使用core.async 中的alts! 来完成此任务 (https://clojure.github.io/core.async/#clojure.core.async/alts!):

(def x (chan 10))

(go (>! x 1)
    (>! x 2)
    (>! x 3))

(defn read-all [from-chan]
  (<!! (go-loop [res []]
           (let [[v _] (alts! [from-chan] :default :complete)]
             (if (= v :complete)
               res
               (recur (conj res v)))))))

(read-all x) 
;; output: [1 2 3]

(read-all x)
;; output: []

(go (>! x 10)
    (>! x 20)
    (>! x 30)
    (>! x 40))

(read-all x)
;; output: [10 20 30 40]

在go-loop 内部，这个(a/alts! [from-chan] :default :complete) 尝试从通道中读取任何值，如果没有可用的值，它会立即发出默认值，因此您会看到您应该打破循环并返回累积值。

更新：由于cljs中没有阻塞读取（<!!），您可以通过以下方式重写：

(defn read-all [from-chan]
  (go-loop [res []]
    (let [[v _] (alts! [from-chan] :default :complete)]
      (if (= v :complete)
        res
        (recur (conj res v)))))))

所以它会返回通道，然后从那里读取一个值：

(go (let [res (<! (read-all x))]
      (println res)
      ;; do something else
      ))

Answer 2

你可以使用clojure.core.async/reduce:

;; demo setup
(def ch (async/chan 2))
(async/>!! ch :foo)
(async/>!! ch :bar)

;; background thread to print reduction result
(async/thread
  (prn (async/<!! (async/reduce conj [] ch))))

;; closing the channel…
(async/close! ch)

;; …terminates the reduction and the result gets printed out:
;; [:foo :bar]

clojure.core.async/reduce 返回一个通道，当原始通道关闭时，该通道将产生一个值。在内部，它使用 go 块，并在从原始通道获取元素之间释放控制权。

如果您想在经过一定时间后生成一个值，无论原始通道是否关闭，您可以将原始通道包装在一个直通通道中，该通道将在超时后自行关闭，或者您可以使用减少步骤的自定义方法（可能是@leetwinski 建议的方法）。

Answer 3

使用into

返回一个包含单个（集合）结果的通道 从通道中取出的项目与提供的集合相连。 ch 必须在 into 产生结果之前关闭。

这样的东西应该可以工作（它应该打印来自events-chan的事件给定的events-chan在完成发布事件后关闭）：

(go
  (println (<! (into [] events-chan))))

源通道需要结束（close），否则不能将所有事件都放入一个集合中。

编辑：

重新阅读您的问题，并不清楚您要完成什么。无论您想做什么，chan->vector 都需要返回一个通道，以便任何调用它的人都可以等待结果。其实chan->vector就是into：

; chan->vector ch:Chan<Event> -> Chan<Vector[Event]>
(defn chan->vector [ch]
  (into [] ch))

(go
  (let [events (<! (chan->vector events-chan))]
    (println events) ; Do whatever with the events vector
    ))

正如我上面提到的，如果事件 chan 永远不会关闭，那么您必须更多地考虑如何使用事件。没有神奇的解决方案。您想按时间间隔对事件进行批处理吗？按事件数量？通过这些的组合？

综上所述，chan->vector 就是into。

Answer 4

虽然在 Clojure 和许多其他语言中可能，但您想要做的事情在 ClojureScript 中是不可能的。

您想要一个在收听频道时阻塞的功能。但是，ClojureScript 的 core.async 版本doesn't include the blocking operators。为什么？因为 ClojureScript 不会阻塞。

我找不到可靠的来源来支持最后一句话。网络上似乎有很多关于这个话题的困惑。不过，我很确定我在说什么，因为 ClojureScript 最终会变成 JavaScript，这就是 JavaScript 的工作原理。

确实，JavaScript never blocks，既不在浏览器上，也不在 in Node.js。为什么？据我了解，它使用单线程，所以如果它被阻塞，用户将无法在浏览器中执行任何操作。

所以不可能做你想做的事。这是设计使然，因为它可能会产生灾难性的用户体验效果。 ClojureScript 通道就像 JavaScript 事件；就像您不希望事件侦听器在等待事件发生时阻塞用户界面一样，您也不应该希望通道在等待新值时阻塞。

请尝试使用一个回调函数，该函数在传递新值时被调用。