检查线程关闭并在关闭后执行任务

Question

我正在创建一个具有 10 个线程的固定大小的线程池，如下所示：

ExecutorService executorService = Executors.newFixedThreadPool(10);

for (int i = 0; i < poolSize; ++i) {
   executorService.execute(factory.get());
}

现在，在产生 10 个线程之后，每个线程都在执行一个可运行的对象。这里的Runnable由factory.get()提供

现在，我想做一件事，即一旦线程关闭，即它完成了一项任务，它就会再次选择另一个可运行对象并开始运行它。因此，基本上检查所有 10 个线程的状态并在线程关闭时再次执行可运行。

我知道我可以做类似的事情，其中​​我可以调用 executorService shutdown() 方法来做到这一点：

   while (!executorService.isShutdown()) {
        try {
            executorService.execute(factory.get());
        } catch (RejectedExecutionException e) {
            if (!executorService.isShutdown())
                log.warn("task submission rejected");
        }
    }

但是在这种方法中，问题是我一直在调用执行函数，这会增加 LinkedBlockingQueue 并且不会达到我的目的。

如果我能以某种方式检查 Threads 状态，那就太好了，因为这样可以避免返回从 executor 服务获取更多工作的开销。

请建议我如何在代码中做到这一点。

Answer 1

您可以使用有界执行器。以下是 code 来自 jcip 的 Brian Goetz

/**
 * BoundedExecutor
 * <p/>
 * Using a Semaphore to throttle task submission
 *
 * @author Brian Goetz and Tim Peierls
 */
public class BoundedExecutor {
    private final Executor exec;
    private final Semaphore semaphore;

    public BoundedExecutor(Executor exec, int bound) {
        this.exec = exec;
        this.semaphore = new Semaphore(bound);
    }

    public void submitTask(final Runnable command)
            throws InterruptedException {
        semaphore.acquire();
        try {
            exec.execute(new Runnable() {
                public void run() {
                    try {
                        command.run();
                    } finally {
                        semaphore.release();
                    }
                }
            });
        } catch (RejectedExecutionException e) {
            semaphore.release();
        }
    }
}

然后创建并使用它：

ExecutorService executorService = Executors.newFixedThreadPool(10);
BoundedExecutor boundedExecutor = new BoundedExecutor(executorService,10);

while (!executorService.isShutdown()) {
    try {
        boundedExecutor.submitTask(factory.get());
    } catch (InterruptedException e) {
    }
}       

这样你将始终有 10 个线程在运行，新任务只有在旧任务完成后才会提交，你可以通过关闭执行器来停止执行。

Answer 2

如果要连续运行 10 个线程，为什么要使用线程池？只需自己启动 10 个线程即可。

为什么在 10 个工作线程之上有一个“控制器”线程？只需让每个线程请求执行下一个任务即可。

以下假设factory.get() 是线程安全的。

for (int i = 0; i < 10; i++) {
    new Thread(() -> {
        for (;;) {
            Runnable r = factory.get(); // <-- must be thread-safe
            if (r == null)
                break; // stop thread if no more tasks to execute
            try {
                r.run();
            } catch (Exception e) {
                // log exception here
            }
        }
    }).start();
}

如果需要，将 10 个线程对象保存在一个数组中。

如果您希望能够在不让factory.get() 返回 null 的情况下关闭线程，请添加一个 volatile 布尔值，并在循环中检查它。