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

Question

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

(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 block 。但是如果我将 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 block 只能同时处理 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"

我知道 async/thread 但它似乎与 (future ...) 相同。

这可能吗？

Answer 1

我建议:

• 使用 futures 创建线程，并让它们使用 put! 从任何 go block 外部将结果放回核心异步 channel ，例如:(future (put! chan ( worker 函数)))
• 然后使用 go block 在该(单个) channel 上等待，在获得结果时将其放入。