我有一个流程需要并行计算许多小任务,然后按任务的自然顺序处理结果。为此,我进行了以下设置:
一个简单的 ExecutorService 和一个阻塞队列,当 Callable 提交给执行程序时,我将使用它来保持返回的 Future 对象:
ExecutorService exec = Executors.newFixedThreadPool(15);
LinkedBlockingQueue<Future<MyTask>> futures = new LinkedBlockingQueue<Future<MyTask>>(15 * 64);
一些调试代码,用于计算提交的数量和已处理的任务数量,并定期将它们写出来(注意 processed
在任务代码本身的末尾递增):
AtomicLong processed = new AtomicLong(0);
AtomicLong submitted = new AtomicLong(0);
Timer statusTimer = new Timer();
statusTimer.schedule(new TimerTask() {
@Override
public void run() {
l.info("Futures: " + futures.size() + "; Submitted: " + submitted.get() + "; Processed: " + processed.get() + "; Diff: " + (submitted.get() - processed.get())));
}
}, 60 * 1000, 60 * 1000);
一个线程从队列(实际上是一个生成器)中获取任务并将它们提交给执行器,将生成的 Future 放入 futures
队列中(这是我确保我不这样做的方式提交太多任务导致内存不足):
Thread submitThread = new Thread(() ->
{
MyTask task;
try {
while ((task = taskQueue.poll()) != null) {
futures.put(exec.submit(task));
submitted.incrementAndGet();
}
} catch (Exception e) {l .error("Unexpected Exception", e);}
}, "SubmitTasks");
submitThread.start();
当前线程然后从
-s完成的任务并处理结果:futures
队列中取出
while (!futures.isEmpty() || submitThread.isAlive()) {
MyTask task = futures.take().get();
//process result
}
当我在具有 8 个内核的服务器上运行它时(请注意,代码当前使用 15 个线程),CPU 利用率峰值仅在 60% 左右。我看到我的调试输出是这样的:
INFO : Futures: 960; Submitted: 1709710114; Processed: 1709709167; Diff: 947
INFO : Futures: 945; Submitted: 1717159751; Processed: 1717158862; Diff: 889
INFO : Futures: 868; Submitted: 1724597808; Processed: 1724596954; Diff: 853
INFO : Futures: 940; Submitted: 1732030120; Processed: 1732029252; Diff: 871
INFO : Futures: 960; Submitted: 1739538576; Processed: 1739537758; Diff: 818
INFO : Futures: 960; Submitted: 1746965761; Processed: 1746964811; Diff: 950
线程转储显示许多线程池线程像这样阻塞:
"pool-1-thread-14" #30 prio=5 os_prio=0 tid=0x00007f25c802c800 nid=0x10b2 waiting on condition [0x00007f26151d5000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x00007f2fbb0001b0> (a java.util.concurrent.locks.ReentrantLock$NonfairSync)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireInterruptibly(AbstractQueuedSynchronizer.java:897)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(AbstractQueuedSynchronizer.java:1222)
at java.util.concurrent.locks.ReentrantLock.lockInterruptibly(ReentrantLock.java:335)
at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:439)
at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
我对调试输出的解释是,在任何给定的时间点,我至少有数百个任务已提交给执行程序服务,但尚未处理(我还可以在堆栈跟踪中确认SubmitTasks 线程在 LinkedBlockingQueue.put
上被阻塞。然而,堆栈跟踪(和服务器利用率统计信息)向我显示执行程序服务在 LinkedBlockingQueue.take 上被阻止(我假设是内部任务队列为空)。
我读错了什么?
最佳答案
2.5 年后,我看到这个问题收到了一些意见,我想我会提供一个跟进。
经过多次更改和测试,我最终将任务分成 10000 个一组(也就是说,每个 Future
负责一组 10000 个 MyTask
任务, 而不仅仅是 1).这样,ExecutorService
每秒执行大约 10-20 个任务(而不是我“要求”它执行的相当高的 100000-200000。这种方法显着提高了速度并导致完全 100% CPU 利用率。
事后看来,每秒执行超过 100k 个任务似乎“不合理”。我的读物是在并发管理/锁定开销和上下文切换(一个猜想)上花费了太多时间。
关于java - 修复了线程池线程阻塞,当提交了足够多的任务时,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/34195954/