java - ThreadPoolExecutor 的 getActiveCount()

Question

我有一个 ThreadPoolExecutor，当我调用 getActiveCount() 时它似乎在欺骗我。然而，我没有做过很多多线程编程，所以也许我做错了什么。

这是我的 TPE

@Override
public void afterPropertiesSet() throws Exception {

    BlockingQueue<Runnable> workQueue;
    int maxQueueLength = threadPoolConfiguration.getMaximumQueueLength();
    if (maxQueueLength == 0) {
        workQueue = new LinkedBlockingQueue<Runnable>();
    } else {
        workQueue = new LinkedBlockingQueue<Runnable>(maxQueueLength);
    }

    pool = new ThreadPoolExecutor(
                   threadPoolConfiguration.getCorePoolSize(),
                   threadPoolConfiguration.getMaximumPoolSize(),
                   threadPoolConfiguration.getKeepAliveTime(),
                   TimeUnit.valueOf(threadPoolConfiguration.getTimeUnit()),
                   workQueue,
                   // Default thread factory creates normal-priority,
                   // non-daemon threads.
                   Executors.defaultThreadFactory(),
                   // Run any rejected task directly in the calling thread.
                   // In this way no records will be lost due to rejection
                   // however, no records will be added to the workQueue
                   // while the calling thread is processing a Task, so set
                   // your queue-size appropriately.
                   //
                   // This also means MaxThreadCount+1 tasks may run
                   // concurrently. If you REALLY want a max of MaxThreadCount
                   // threads don't use this.
                   new ThreadPoolExecutor.CallerRunsPolicy());
}

在这个类中，我还有一个 DAO，我将其传递到我的 Runnable (FooWorker) 中，如下所示:

@Override
public void addTask(FooRecord record) {
    if (pool == null) {
        throw new FooException(ERROR_THREAD_POOL_CONFIGURATION_NOT_SET);
    }
    pool.execute(new FooWorker(context, calculator, dao, record));
}

FooWorker 通过状态机通过 calculator 运行 record(唯一的非单例)，然后通过 将转换发送到数据库>dao，像这样:

public void run() {
    calculator.calculate(record);
    dao.save(record);
}

一旦我的主线程完成创建新任务，我会尝试并等待以确保所有线程都成功完成:

while (pool.getActiveCount() > 0) {
    recordHandler.awaitTermination(terminationTimeout, 
                                   terminationTimeoutUnit);
}

我从输出日志中看到的(由于线程可能不可靠)是 getActiveCount() 过早返回零，而 while() 循环正在退出，而我的最后一个线程仍在打印来自 计算器。

请注意，我也曾尝试调用 pool.shutdown()，然后使用 awaitTermination，但下次我的作业运行时，池仍然关闭。

我唯一的猜测是在线程内部，当我将数据发送到dao时(因为它是Spring在主线程中创建的单例...)， java 正在考虑线程不活动，因为(我假设)它正在处理/等待主线程。

凭直觉，仅根据我所看到的，这是我的猜测。但是……真的是这样吗？有没有一种方法可以“正确执行”，而无需在 run() 的顶部放置一个手动递增的变量，并在末尾放置一个递减的变量来跟踪线程数？

如果答案是“不要传入 dao”，那么我是否必须为每个线程“新建”一个 DAO？我的过程已经是一个(漂亮、高效)的野兽，但这真的很糟糕。

Answer 1

作为the JavaDoc of getActiveCount states ，这是一个近似值:您不应以此作为任何主要业务逻辑决策的基础。

如果您想等待所有 计划任务完成，那么您应该简单地使用

pool.shutdown();
pool.awaitTermination(terminationTimeout, terminationTimeoutUnit);

如果您需要等待特定 任务完成，您应该使用submit()而不是 execute() 然后检查 Future完成对象(如果您想非阻塞地使用 isDone()，或者简单地调用 get()，它会阻塞直到任务完成)。