java - 使用corePoolSize而不是maxPoolSize的InvokeAll方法

Question

我有一个需要将数据提取到ES的应用程序。有多个行，因此我使用ThreadPoolTaskExecutor并行处理了每批50个数据的摄取。此外，我在使用者中获取了这些数据，并且在处理所有记录之前不希望确认，因此我封装了任务在Runnable类中，我将它们添加到可调用对象列表中，并在列表包含所有数据时使用invokeAll()。
我的ThreadPoolTaskExecutor的maxPoolSize是200，而corePoolSize是50。因此，我期望应用程序一次使用200线程，因为据我了解，当排队的任务少于ThreadPoolTaskExecutor时corePoolSize使用高达corePoolSize的线程，后来它开始将线程添加到maxPoolSize并执行任务。当我检查日志和visualVm时，发现只有50线程正在执行，因此，我将corePoolSize增加到70，并发现70线程正在执行。其中，队列的大小应大于3000，而列表的大小应大于30万。该问题的示例代码。

public void Test() throws InterruptedException {
List<Callable<Object>> tagList = new ArrayList<Callable<Object>>(1000);
for(int i=0; i<400; i++){
  tagList.add(Executors.callable(new Runnable() {
   @Override
   public void run() {
     System.out.println("Thread "+ Thread.currentThread());
     try {
       Thread.sleep(4000);
     } catch (InterruptedException e) {
       e.printStackTrace();
     }
   }
 }));
}
ThreadPoolTaskExecutor.getThreadPoolExecutor().invokeAll(tagList);
}

信息| 2021-02-14 00:06:32.086 + 05:30 |申请| 134 |当前线程数:50总线程数:200
信息| 2021-02-14 00:06:33.085 + 05:30 |申请| 134 |当前线程数:50总线程数:200
请有人解释我是否缺少什么。

Answer 1

maxPoolSize of my ThreadPoolTaskExecutor is 200 and corePoolSize is50.So,I was expecting the application to use 200 threads at one time as by my understanding ThreadPoolTaskExecutor uses threads upto tocorePoolSize when queued up tasks are less than the corePoolSize andlater it starts adding threads upto the maxPoolSize and executes thetasks.

那不是很正确。与 ThreadPoolExecutor一样:

When a new task is submitted (..), and fewer than corePoolSize threads are running, a new thread is created to handle the request, even if other worker threads are idle.If there are more than corePoolSize but less than maximumPoolSizethreads running, a new thread will be created only if the queue isfull.

因此，即使maxPoolSize为 200，您也只能运行50个线程的原因。
从 ThreadPoolTaskExecutor可以读取:

The default configuration is a core pool size of 1, with unlimited maxpool size and unlimited queue capacity. This is roughly equivalent toExecutors.newSingleThreadExecutor(), sharing a single thread for alltasks. Setting "queueCapacity" to 0 mimicsExecutors.newCachedThreadPool(), with immediate scaling of threads inthe pool to a potentially very high number. Consider also setting a"maxPoolSize" at that point, as well as possibly a higher"corePoolSize" (see also the "allowCoreThreadTimeOut" mode ofscaling).

看来您必须将 queueCapacity设置为零才能满足您的要求。