如何防止关闭！-ing 在放入 on-chan 之前

Question

我想运行类似的代码

(->> input
     (partition-all 5)
     (map a-side-effect)
     dorun)

异步划分输入和输出（副作用）。

然后我编写了下面的代码进行实验。

;; using boot-clj
(set-env! :dependencies '[[org.clojure/core.async "0.2.374"]])
(require '[clojure.core.async :as async :refer [<! <!! >! >!!]])

(let [input (range 18)
      c (async/chan 1 (comp (partition-all 5)
                            (map prn)))]
  (async/onto-chan c input false)
  (async/close! c))

此代码的解释：

• 实际上输入中的元素及其数量在运行之前并没有定义，输入中的元素可以取0到10之间的一些数字。
• async/onto-chan 用于将元素的 Seq（输入的片段）放入通道 c 并将被多次调用，因此第三个参数是 false。
• prn 是 a-side-effect 的替代品。

我希望上面的代码打印出来

[0 1 2 3 4]
[5 6 7 8 9]
[10 11 12 13 14]
[15 16 17]

在 REPL 中，它不打印任何字符。

然后我添加一个等待的时间，像这样

(let [c (async/chan 1 (comp (partition-all 5)
                            (map prn)))]
  (async/onto-chan c (range 18) false)
  (Thread/sleep 1000) ;wait
  (async/close! c))

这段代码在上面给出了我预期的输出。

然后我检查core.async/onto-chan。

我认为发生了什么：

1. 频道 c 在我的代码中是 core.async/close!ed。
2. core.async/onto-chan 参数的每一项都被放在onto-chan 中的go-loop 中（core.async/>!）徒劳无功，因为通道c 已关闭。

有没有办法将项目放在close!ing 之前？ 写一个同步版本的onto-chan 不使用go-loop？

还是我的想法错了？

Answer 1

您使用Thread.sleep 的第二个示例只是错误地“起作用”。

它起作用的原因是，从c的转换器出来的每个转换结果值都是nil，并且由于通道中不允许nils，因此抛出异常，并且不放置任何值进入c：这是允许生产者onto-chan 继续放入频道而不阻塞等待的原因。如果您将第二个示例粘贴到 REPL 中，您将看到四个堆栈跟踪 - 每个分区一个。

nils 当然是由于映射到prn，这是一个副作用函数，它为所有输入返回nil。

如果我正确理解你的设计，你的目标是做这样的事情：

(defn go-run! [ch proc]
  (async/go-loop []
    (when-let [value (<! ch)]
      (proc value)
      (recur))))

(let [input (range 18)
      c (async/chan 1 (partition-all 5))]
  (async/onto-chan c input)
  (<!! (go-run! c prn)))

• 您确实需要生产者和消费者，否则您的程序将阻塞。我介绍了一个go-loop 消费者。
• 一般来说，map 和副作用不能很好地结合在一起，所以我将副作用 prn 提取到消费者中。
• onto-chan 不能被“多次”调用（至少在显示的代码中），因此它不需要 false 参数。

Answer 2

采用 megakorre 的想法：

(let [c (async/chan 1 (comp (partition-all 5)
                            (map prn)))
      put-ch (async/onto-chan c (range 18) false)]
  (async/alts!! [put-ch])
  (async/close! c))