Clojure async/go 如何停放阻塞代码

Question

我使用一些 Java 库来发出非异步获取和发布请求。我曾经将此类请求包装到期货中，它为我解决了“等待问题”（我的意思是等待响应）

(defn unchangeable-lib-request [n]
  (Thread/sleep 1000)
  n)

(defn process [n]
  (let [res (atom [])]
    (dotimes [i n]
      (future (swap! res conj (unchangeable-lib-request i))))
    (loop []
      (if (> n (count @res))
        (recur)
        @res))))

(time (process 9))

;; "Elapsed time: 1000.639079 msecs"
;; => [8 7 5 6 4 3 2 1 0]

但是我需要创建数百个请求，这会产生性能问题。我发现了 core.async 和 go 块。但是如果我将 go-blocks 与这个库一起使用，它不会解决“等待问题”

(defn unchangeable-lib-request [n]
  (Thread/sleep 1000)
  n)

(defn process [n]
  (let [c (async/chan 10)]
    (dotimes [i n]
      (async/go
        (async/>! c (unchangeable-lib-request i))))
  (loop [result []]
    (if (> n (count result))
      (recur (conj result (async/<!! c)))
      result))))

(time (process 9))

;; "Elapsed time: 2001.770183 msecs"
;; => [0 4 1 6 7 2 5 3 8]

Go 块只能同时处理 8 个请求。是否有可能编写一些异步包装器来停放 go-block 并提供异步发出 100 个请求而不会相互阻塞的能力？

(defn process [n]
  (let [c (async/chan 10)]
    (dotimes [i n]
      (async/go
        (async/>! c (magic-async-parking-wrapper
                      (unchangeable-lib-request i))))
  (loop [result []]
    (if (> n (count result))
      (recur (conj result (async/<!! c)))
      result))))

(time (process 9))

;; "Elapsed time: 1003.2563 msecs"

我知道异步/线程，但这似乎与（未来...）相同。

有可能吗？

Answer 1

我建议：

• 使用期货创建线程，并让他们使用put!将结果从任何go块外部放回核心异步通道，例如：(future (put! chan (worker-function)))
• 然后使用 go 块在该（单个）通道上等待，在获得结果时输入结果。

Answer 2

这是你使用clojure.core.async/pipeline-blocking的地方

(require '[clojure.core.async :as a :refer [chan pipeline-blocking]])


(let [output-chan (chan 100)
      input-chan (chan 1000)]
  (pipeline-blocking 4 ; parallelism knob
                     output-chan 
                     (map unchangeable-lib-request) 
                     input-chan)
  ;; Consume results from output-chan, put operations on input-chan
  [output-chan input-chan]
  )

这会产生 n 个（在本例中为 4）个线程，它们一直忙于执行 unchangeable-lib-request。

使用output-chan 的缓冲区大小来微调您希望提前发生多少请求。

使用input-chan 的缓冲区大小来微调您希望在没有反向传播的情况下调度多少请求（阻塞input-chan）。