Java 删除 ThreadPoolExecutor 的所有排队任务

Removing all queued tasks of an ThreadPoolExecutor

提问by Malax

i have this rather simple question about the ThreadPoolExecutor. I have the following situation: I have to consume objects from a queue, create the appropiate worker tasks for them and submit them to the ThreadPoolExecutor. This is quite simple. But within a shutdown scenario manyworkers may be queued to execution. Since one of those tasks might be running for an hour, and i want a relativly fast graceful shutdown of the application i want to discard all queued tasks from the ThreadPoolExecutor while the already processing tasks should be completed normally.

我有一个关于ThreadPoolExecutor 的相当简单的问题。我有以下情况：我必须从队列中使用对象，为它们创建适当的工作任务并将它们提交给 ThreadPoolExecutor。这很简单。但是在关闭场景中，许多工作人员可能会排队等待执行。由于其中一个任务可能会运行一个小时，并且我想要相对快速地正常关闭应用程序，我想从 ThreadPoolExecutor 丢弃所有排队的任务，而已经处理的任务应该正常完成。

The ThreadPoolExecutor documentation has a remove()method but only allows specific tasks to be removed. purge()only works for already canceled Future tasks. My idea was to clear the queue holding all queued tasks. The ThreadPoolExecutor provides access to this internal queue but the documentation states:

ThreadPoolExecutor 文档有一个remove()方法，但只允许删除特定任务。purge()仅适用于已取消的 Future 任务。我的想法是清除保存所有排队任务的队列。ThreadPoolExecutor 提供对这个内部队列的访问，但文档指出：

Method getQueue() allows access to the work queue for purposes of monitoring and debugging. Use of this method for any other purpose is strongly discouraged.

方法 getQueue() 允许访问工作队列以进行监视和调试。强烈建议不要将此方法用于任何其他目的。

So grabbing this queue and clearing it is not an option. Also, this snippet of the documentation says:

所以抓住这个队列并清除它不是一种选择。此外，文档的这个片段说：

Two supplied methods, remove(java.lang.Runnable) and purge() are available to assist in storage reclamation when large numbers of queued tasks become cancelled.

提供的两个方法 remove(java.lang.Runnable) 和 purge() 可用于在大量排队任务被取消时协助存储回收。

How? Sure, i can maintain a list of all tasks i submitted to the executor and in a shutdown case i iterate over all entries and remove them from the ThreadPoolExecutor with the remove() method... but... come on, this is a waste of memory and a hassle to maintain this list. (Removing already executed tasks for example)

如何？当然，我可以维护我提交给执行程序的所有任务的列表，在关闭的情况下，我遍历所有条目并使用 remove() 方法将它们从 ThreadPoolExecutor 中删除......但是......来吧，这是一个浪费内存，维护这个列表很麻烦。（例如删除已执行的任务）

I appreciate any hints or solutions!

我感谢任何提示或解决方案！

采纳答案by Sbodd

Have you considered wrapping the ExecutorService? Create a

您是否考虑过包装 ExecutorService？创建一个

CleanShutdownExecutorService implements Executor 

that delegates all calls to another Executor, but keeps the Futures in a list of its own. CleanShutdownExecutorService can then have a cancelRemainingTasks() method that calls shutdown(), then calls cancel(false) on all the Futures in its list.

将所有调用委托给另一个 Executor，但将 Futures 保留在自己的列表中。然后 CleanShutdownExecutorService 可以有一个 cancelRemainingTasks() 方法，该方法调用 shutdown()，然后在其列表中的所有 Future 上调用 cancel(false)。

回答by Bombe

As ExecutorService.shutdown()is not doing enough and ExecutorService.shutdownNow()is doing too much I guess you have to write up something in the middle: remember all your submitted tasks and remove them manually after (or before) calling shutdown().

由于ExecutorService.shutdown()做得不够，而ExecutorService.shutdownNow()做得太多，我猜你必须在中间写一些东西：记住所有提交的任务并在调用shutdown().

回答by Max A.

Bombe's answer is exactly what you want. shutdownNow()stops everything using the nuke and paveapproach. This is the best thing you can do, short of subclassing the implementation of ThreadPoolExecutorthat you're using.

Bombe 的回答正是您想要的。shutdownNow()使用核武器和铺路方法停止一切。这是您可以做的最好的事情，除了对ThreadPoolExecutor您正在使用的实现进行子类化之外。

回答by jitter

A crazy and unclean solution which might work (not real thought through or tested) would be to overwrite the interrupt()of your WorkerTasks which only in case some global value is set refuse to shutdown when interrupt()is called on them by shutdownNow().

一个可能有效的疯狂和不干净的解决方案（不是真正经过深思熟虑或测试）是覆盖interrupt()您的 WorkerTasks 的 ，只有在某些全局值被设置为拒绝关闭时interrupt()，它们才会被 shutdownNow() 调用。

That should allow you to use shutdownNow()no?

那应该允许您使用shutdownNow()否？

回答by Carl

Tell your thread pool to shutdown, getQueue, for-each the result into individual Runnables, remove each Runnable using the remove method. Depending on the type of queue, you might be able to halt the removes early based on return values.

告诉您的线程池关闭、getQueue、for-each 结果到单独的 Runnable 中，使用 remove 方法删除每个 Runnable。根据队列的类型，您可能能够根据返回值提前停止删除。

Basically, this is grabbing the queue and clearing it, only clearing via the methods that work. Instead of manually remembering all the submissions, you use the fact the thread pool already has to remember all the submissions. However, you will probably need to make a defensive copy of the queue, as I think it's a live view, and therefore removing would probably cause a concurrent modification exception if you were iterating/for-eaching over the live view.

基本上，这是抓取队列并清除它，只能通过有效的方法清除。您不必手动记住所有提交，而是使用线程池已经必须记住所有提交的事实。但是，您可能需要制作队列的防御性副本，因为我认为它是实时视图，因此如果您在实时视图上迭代/for-each，删除可能会导致并发修改异常。

回答by ZZ Coder

I used to work on an app with long running threads. We do this at shutdown,

我曾经在一个具有长时间运行线程的应用程序上工作。我们在关机时这样做，

BlockingQueue<Runnable> queue = threadPool.getQueue();
List<Runnable> list = new ArrayList<Runnable>();
int tasks = queue.drainTo(list);

The list is saved to a file. On startup, the list is added back to the pool so we don't lose any jobs.

该列表将保存到文件中。在启动时，该列表被添加回池中，因此我们不会丢失任何工作。

回答by laura

Doesn't awaitTermination(long timeout, TimeUnit unit)work after shutdown?

awaitTermination(long timeout, TimeUnit unit)关机后不能用？

executor.shutdown(); executor.awaitTermination(60, TimeUnit.SECONDS)

executor.shutdown(); executor.awaitTermination(60, TimeUnit.SECONDS)

回答by iman

You can try allowCoreThreadTimeOut(true);

你可以试试 allowCoreThreadTimeOut(true);

回答by Robert Tupelo-Schneck

This is an old question, but in case this helps somebody else: you could set a volatile boolean when you call shutdown(), and have each submitted task terminate if that boolean is set before really starting. This will allow tasks which have genuinely started to complete, but will prevent queued tasks from starting their actual activity.

这是一个老问题，但如果这对其他人有帮助：您可以在调用 shutdown() 时设置一个 volatile 布尔值，并且如果在真正开始之前设置了该布尔值，则每个提交的任务都会终止。这将允许真正开始完成的任务，但会阻止排队的任务开始它们的实际活动。

回答by zenn1337

You could create your own task queue and pass it to ThreadPoolExecutorconstructor:

您可以创建自己的任务队列并将其传递给ThreadPoolExecutor构造函数：

int poolSize = 1; // number of threads
BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>();
Executor executor = new ThreadPoolExecutor(poolSize, poolSize, 0L, TimeUnit.MILLISECONDS, queue);

When you clear the queue somewhere in your code then the remaining tasks will not be executed:

当您清除代码中某处的队列时，将不会执行剩余的任务：

queue.clear();