我在使用
ThreadPoolExecutor
时遇到一些问题,我无法弄清楚。
我尝试使用两个List<CompletableFuture<Void>> taskList
列表来存储任务,然后提交到线程池,但所有线程都被阻止并且无法处理新请求。代码如下:
有一个线程池。
public class AsyncInvoker {
public class AsyncInvoker {
private ExecutorService excutePool;
public AsyncInvoker(int thread, int queue) {
excutePool= new ThreadPoolExecutor(thread, thread,
0, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(queue), new ThreadPoolExecutor.AbortPolicy());
}
public InvokeCollection newCollection() {
return new InvokeCollection();
}
public class InvokeCollection {
private List<CompletableFuture<Void>> taskList = new ArrayList<>();
public void add(Runnable task) {
CompletableFuture<Void> future = CompletableFuture.runAsync(task, excutePool);
taskList.add(future);
}
public void invokeAll() {
try {
CompletableFuture<Void> future = CompletableFuture.allOf(taskList.toArray(new CompletableFuture[0]));
future.get();
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
我提交的工作是这样的:
@RestController
public class TestController {
AsyncInvoker asyncInvoker = new AsyncInvoker(10,1000000);
@GetMapping("test")
public void test(){
System.out.println("request=======================");
AsyncInvoker.InvokeCollection outer = asyncInvoker.newCollection();
for (int i = 0; i < 10; i++) {
outer.add(()->{
AsyncInvoker.InvokeCollection inner = asyncInvoker.newCollection();
for (int j = 0; j < 100; j++) {
inner.add(()->{
try {
Thread.sleep(100);
System.out.println("===");
}catch (Exception e) {
e.printStackTrace();
}
});
}
inner.invokeAll();
});
System.out.println("i " + i);
}
outer.invokeAll();
}
}
当我多次请求
TestController
时,idea的控制台中有几个===
打印,当我重复请求时,不再有===
打印。我认为线程被阻塞了。线程转储如下所示:
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x000000076c9923a0> (a java.util.concurrent.CompletableFuture$Signaller)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.CompletableFuture$Signaller.block(CompletableFuture.java:1707)
at java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3334)
at java.util.concurrent.CompletableFuture.waitingGet(CompletableFuture.java:1742)
at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)
at com.aa.AsyncInvoker$InvokeCollection.invokeAll(AsyncInvoker.java:59)
然后,如果只有一个
AsyncInvoker.InvokeCollection
而不是 outer
和 inner
,并且我将所有任务添加到一个集合中,效果很好。
而且,如果我使用
private ForkJoinPool forkJoinPool = new ForkJoinPool(thread,new SafeForkJoinWorkerThreadFactory(), null, false);
来处理任务,看起来也不错。
所以我不知道为什么上面写的代码不起作用,是不是线程被阻塞了,为什么?以及为什么在这种情况下
ForkJoinPool
比 ThreadPoolExecutor
更好。
有人可以给我一些帮助吗?非常感谢!!
问题是,您的
asyncInvoker
实例中的线程池执行器最多有 10 个可用线程。除非线程池执行器线程完成其任务,否则它无法用于任何其他任务执行。为了完成 outer
的任务,您需要 1 个线程来执行该任务,并需要 100 个线程来完成 inner
的所有任务。为了完成 inner
的任务,必须完成该实例中的所有 100 项任务。但是inner
的第10个任务将不再被执行,因为执行器中的所有线程都已经在处理任务了。
使用
ForkJoinPool
时的行为有所不同,因为这使用了工作窃取。 future.get();
不会等待线程执行,而是会搜索其他要做的工作,除非 future
完成。