了解 Clojure 与 ClojureScript 中的 core.async 合并

Question

我正在 Clojure 和 ClojureScript 上尝试使用 core.async，以尝试了解 merge 的工作原理。特别是，merge 是否使输入通道上的任何值可用于立即在合并通道上。

我有以下代码：

(ns async-merge-example.core
  (:require
   #?(:clj [clojure.core.async :as async] :cljs [cljs.core.async :as async])
   [async-merge-example.exec :as exec]))

(defn async-fn-timeout
  [v]
  (async/go
    (async/<! (async/timeout (rand-int 5000)))
    v))

(defn async-fn-exec
  [v]
  (exec/exec "sh" "-c" (str "sleep " (rand-int 5) "; echo " v ";")))

(defn merge-and-print-results
  [seq async-fn]
  (let [chans (async/merge (map async-fn seq))]
    (async/go
      (while (when-let [v (async/<! chans)]
               (prn v)
               v)))))

当我尝试 async-fn-timeout 和大号 seq 时：

(merge-and-print-results (range 20) async-fn-timeout)

对于 Clojure 和 ClojureScript，我都得到了我期望的结果，例如，结果几乎立即开始打印，并带有预期的延迟。

但是，当我尝试 async-fn-exec 与相同的 seq 时：

(merge-and-print-results (range 20) async-fn-exec)

对于 ClojureScript，我得到了我期望的结果，因为结果几乎立即开始打印，并带有预期的延迟。然而对于 Clojure，即使 sh 进程是同时执行的（取决于 core.async 线程池的大小），结果似乎最初是延迟的，然后大部分是一次打印出来的！我可以通过增加 seq 的大小来使这种差异更加明显，例如(range 40)

由于 async-fn-timeout 的结果在 Clojure 和 ClojureScript 上都符合预期，因此要指出 exec 的 Clojure 和 ClojureScript 实现之间的差异..

但我不知道为什么这种差异会导致这个问题？

注意事项：

• 这些观察结果是在 Windows 10 上的 WSL 中进行的
• async-merge-example.exec的源代码如下
• 在 exec 中，由于 Clojure/Java 和 ClojureScript/NodeJS 之间的差异，Clojure 和 ClojureScript 的实现有所不同。

(ns async-merge-example.exec
  (:require
   #?(:clj [clojure.core.async :as async] :cljs [cljs.core.async :as async])))

; cljs implementation based on https://gist.github.com/frankhenderson/d60471e64faec9e2158c

; clj implementation based on https://stackoverflow.com/questions/45292625/how-to-perform-non-blocking-reading-stdout-from-a-subprocess-in-clojure

#?(:cljs (def spawn (.-spawn (js/require "child_process"))))

#?(:cljs
   (defn exec-chan
     "spawns a child process for cmd with args. routes stdout, stderr, and
      the exit code to a channel. returns the channel immediately."
     [cmd args]
     (let [c (async/chan), p (spawn cmd (if args (clj->js args) (clj->js [])))]
       (.on (.-stdout p) "data"  #(async/put! c [:out  (str %)]))
       (.on (.-stderr p) "data"  #(async/put! c [:err  (str %)]))
       (.on p            "close" #(async/put! c [:exit (str %)]))
       c)))

#?(:clj
   (defn exec-chan
     "spawns a child process for cmd with args. routes stdout, stderr, and
      the exit code to a channel. returns the channel immediately."
     [cmd args]
     (let [c (async/chan)]
       (async/go
         (let [builder (ProcessBuilder. (into-array String (cons cmd (map str args))))
               process (.start builder)]
           (with-open [reader (clojure.java.io/reader (.getInputStream process))
                       err-reader (clojure.java.io/reader (.getErrorStream process))]
             (loop []
               (let [line (.readLine ^java.io.BufferedReader reader)
                     err (.readLine ^java.io.BufferedReader err-reader)]
                 (if (or line err)
                   (do (when line (async/>! c [:out line]))
                       (when err (async/>! c [:err err]))
                       (recur))
                   (do
                     (.waitFor process)
                     (async/>! c [:exit (.exitValue process)]))))))))
       c)))

(defn exec
  "executes cmd with args. returns a channel immediately which
   will eventually receive a result map of 
   {:out [stdout-lines] :err [stderr-lines] :exit [exit-code]}"
  [cmd & args]
  (let [c (exec-chan cmd args)]
    (async/go (loop [output (async/<! c) result {}]
                (if (= :exit (first output))
                  (assoc result :exit (second output))
                  (recur (async/<! c) (update result (first output) #(conj (or % []) (second output)))))))))

Answer 1

您的 Clojure 实现在单个线程中使用阻塞 IO。您首先从 stdout 读取，然后在循环中读取 stderr。两者都会阻塞readLine，因此只有在他们真正读完一行后才会返回。因此，除非您的进程向 stdout 和 stderr 创建相同数量的输出，否则一个流最终会阻塞另一个流。

一旦进程完成，readLine 将不再阻塞，一旦缓冲区为空，就会返回nil。所以循环只是完成了缓冲输出的读取，然后最终完成了对“一次性”消息的解释。

您可能想要启动第二个线程来处理从 stderr 读取。

node 不会阻塞 IO，所以默认情况下一切都是异步发生的，一个流不会阻塞另一个。