multithreading - do block 中作业数量的上限？

Question

这是代码:

(ns typedclj.async
  (:require [clojure.core.async
             :as a
             :refer [>! <! >!! <!!
                     go chan buffer
                     close! thread
                     alts! alts!! timeout]]
            [clj-http.client :as -cc]))


(time (dorun
        (let [c (chan)]
          (doseq [i (range 10 1e4)]
            (go (>! c i))))))

我得到了一个错误:

Exception in thread "async-dispatch-12" java.lang.AssertionError: Assert failed: No more than 1024 pending puts are allowed on a single channel. Consider using a windowed buffer.
(< (.size puts) impl/MAX-QUEUE-SIZE)
    at clojure.core.async.impl.channels.ManyToManyChannel.put_BANG_(channels.clj:150)
    at clojure.core.async.impl.ioc_macros$put_BANG_.invoke(ioc_macros.clj:959)
    at typedclj.async$eval11807$fn__11816$state_machine__6185__auto____11817$fn__11819.invoke(async.clj:19)
    at typedclj.async$eval11807$fn__11816$state_machine__6185__auto____11817.invoke(async.clj:19)
    at clojure.core.async.impl.ioc_macros$run_state_machine.invoke(ioc_macros.clj:940)
    at clojure.core.async.impl.ioc_macros$run_state_machine_wrapped.invoke(ioc_macros.clj:944)
    at typedclj.async$eval11807$fn__11816.invoke(async.clj:19)
    at clojure.lang.AFn.run(AFn.java:22)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)...

根据 http://martintrojer.github.io/clojure/2013/07/07/coreasync-and-blocking-io/

... This will break the 1 job = 1 thread knot, thus this thread parking will allow us to scale the number of jobs way beyond any thread limit on the platform (usually around 1000 on the JVM).

core.async gives (blocking) channels and a new (unbounded) thread pool when using 'thread'. This (in effect) is just some sugar over using java threads (or clojure futures) and BlockingQueues from java.util.concurrent. The main feature is go blocks in which threads can be parked and resumed on the (potentially) blocking calls dealing with core.async's channels...

1e4职位已经太多了吗？那么上限是多少呢？

Answer 1

我通常不会像这样咆哮，所以我希望你能原谅我这个过错:

在一个更完美的世界中，每个程序员都会在 sleep 前和醒来时的第一件事中对自己重复五次“没有无限队列这样的东西”。这种思维模式需要弄清楚系统中将如何处理背压，以便当过程中某处出现减速时，之前的部件有办法找出它并降低自己的响应速度。在 core.async 中，默认背压是即时的，因为默认缓冲区大小为零。在有人准备消费它之前，没有 go block 成功地将某些东西放入 chan 中。

chans 看起来基本上是这样的:

"queue of pending puts" --> buffer --> "queue of pending takes"

putter 和 taker 队列旨在为通过此管道通信的两个进程留出时间来安排自己，以便取得进展。没有这些，线程就没有空间进行调度，并且会发生死锁。它们不打算用作缓冲区。这就是中间缓冲区的用途，这就是使其成为唯一具有明确大小的缓冲区的设计。通过在 chan 中设置缓冲区的大小，为您的系统显式设置缓冲区大小:

user> (time (dorun
        (let [c (chan 1e6)]
          (doseq [i (range 10 1e4)]
            (go (>! c i))))))
"Elapsed time: 83.526679 msecs"
nil

在这种情况下，我已经“计算出”如果有多达一百万个等待工作，我的系统作为一个整体将处于良好状态。当然，您在现实世界中的体验会有所不同，而且对于您的情况非常独特。

谢谢你的耐心，