multithreading - 为什么在调用Clojure中的Future时仅运行32个线程？

Question

在对 future 进行测试时，我发现:

user=> (time (doall (map deref (for [i (range 1000)]
  #_=>    (future (Thread/sleep 1000))))))
"Elapsed time: 32058.208 msecs"

当看到2的幂时，我的头会响起警铃。闻起来好像只有32个线程启动了。

一些针对性的实验:

user=> (time (doall (map deref (for [i (range 32)]
  #_=> (future (Thread/sleep 1000))))))
"Elapsed time: 1002.302 msecs"
(nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil)
user=> (time (doall (map deref (for [i (range 64)]
  #_=> (future (Thread/sleep 1000))))))
"Elapsed time: 2004.24 msecs"
(nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil)
user=> (time (doall (map deref (for [i (range 65)]
  #_=> (future (Thread/sleep 1000))))))
"Elapsed time: 3005.279 msecs"
(nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil     nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil nil)

确认这一点。对于32个 future 任务的每一部分，我们都会看到额外的一秒。

创造 future 的代码是

(defmacro future
  "Takes a body of expressions and yields a future object that will
  invoke the body in another thread, and will cache the result and
  return it on all subsequent calls to deref/@. If the computation has
  not yet finished, calls to deref/@ will block, unless the variant of
  deref with timeout is used. See also - realized?."
  {:added "1.1"}
  [& body] `(future-call (^{:once true} fn* [] ~@body)))

future 通话中有趣的一点是

    fut (.submit clojure.lang.Agent/soloExecutor ^Callable f)]

转换为:

volatile public static ExecutorService soloExecutor = Executors.newCachedThreadPool(
    createThreadFactory("clojure-agent-send-off-pool-%d", sendOffThreadPoolCounter));

因此，这确实创建了一个无限制的线程池。

那么，为什么只运行32个线程而不是创建1000个线程并在一秒钟内返回呢？

Answer 1

这

(time (doall (map deref (for [i (range 65)] (future (Thread/sleep 1000))))))
; "Elapsed time: 3005.244983 msecs"

实际上应该花费65秒，因为您要在主线程上依次取消对 future 的引用。当refref出现时，这些 future 甚至还没有开始，因为 for是惰性的。之所以得到32的倍数，是 range的分块行为。

与非分块版本比较

 (time (doall (map deref (for [i (apply list (range 65))] (future (Thread/sleep 1000))))))
; "Elapsed time: 64997.260808 msecs"

正如Rörd在评论中所指出的那样，添加另一个 doall以消除将来创作的懒惰可以解决该问题。

(time (doall (map deref (doall (for [i (range 65)] (future (Thread/sleep 1000)))))))
; "Elapsed time: 999.263631 msecs"

测试所有 future 完成情况的另一种方法是等待直到最后一个 future 完成才兑现的 promise 。

(defn test-futures [n]
  (let [a (atom 0)
        p (promise)
        f (fn [] (swap! a inc) (when (= @a n) (deliver p n)))]
    (doseq [i (range n)] (future (do (Thread/sleep 1000) (f))))
    (deref p))) 

现在，您可以看到这些 future 中的64、65或1000个 future 的完成大约在1秒钟内完成。

(time (test-futures 64))
; "Elapsed time: 996.262272 msecs"
; 64
(time (test-futures 65))
; "Elapsed time: 996.554436 msecs"
; 65
(time (test-futures 1000))
; "Elapsed time: 1000.247374 msecs"
; 1000