ThreadPoolExecutor修复了具有自定义行为的线程池

Question

我不熟悉这个话题。。。我正在使用由Executors创建的ThreadPoolExecutor。新的FixedThreadPool（10），池满后，我开始得到一个RejectedExecutionException。有没有办法“强制”执行者将新任务置于“等待”状态，而不是拒绝它并在池释放时启动它？

谢啦

与此相关的问题https://github.com/evilsocket/dsploit/issues/159

涉及的代码行https://github.com/evilsocket/dsploit/blob/master/src/it/evilsocket/dsploit/net/NetworkDiscovery.java#L150

Answer 1

基于前面的建议，您可以使用阻塞队列构造一个固定大小的ThreadPoolExecutor。如果随后提供自己的拒绝执行处理程序（RejectedExecutionHandler），将任务添加到阻塞队列中，则其行为将如所述。

下面是一个如何构建这样一个执行者的示例：

int corePoolSize    = 10;
int maximumPoolSize = 10;
int keepAliveTime   =  0;
int maxWaitingTasks = 10;

ThreadPoolExecutor blockingThreadPoolExecutor = new ThreadPoolExecutor(
        corePoolSize, maximumPoolSize,
        keepAliveTime, TimeUnit.SECONDS,
        new ArrayBlockingQueue<Runnable>(maxWaitingTasks),
        new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                try {
                    executor.getQueue().put(r);
                } catch (InterruptedException e) {
                    throw new RuntimeException("Interrupted while submitting task", e);
                }
            }
        });

Answer 2

线程很有可能调用export，将mStoped设置为false并关闭执行程序，但是：

• 正在运行的线程可能处于while（！mStopped）循环的中间，并尝试向执行器提交任务，该执行器已在退出时关闭

我建议：

• 使易挥发
• 处理执行器在循环中间关闭的情况（例如，通过捕获RejectedExecutionException，或者更好：在while循环之后关闭执行器，而不是在exit方法中关闭）

Answer 3

如果使用执行器。newFixedThreadPool（10） 它对任务进行排队，等待线程就绪。

此方法为

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

如您所见，使用的队列是无界的（这本身可能是个问题），但这意味着队列永远不会填满，您永远不会被拒绝。

顺便说一句：如果您有CPU绑定的任务，可以选择最佳线程数

int processors = Runtime.getRuntime().availableProcessors();
ExecutorService es = Executors.newFixedThreadPool(processors);

一个可以说明这种情况的测试类

public static void main(String... args) {
    ExecutorService es = Executors.newFixedThreadPool(2);
    for (int i = 0; i < 1000 * 1000; i++)
        es.submit(new SleepOneSecond());

    System.out.println("Queue length " + ((ThreadPoolExecutor) es).getQueue().size());
    es.shutdown();
    System.out.println("After shutdown");
    try {
        es.submit(new SleepOneSecond());
    } catch (Exception e) {
        e.printStackTrace(System.out);
    }
}

static class SleepOneSecond implements Callable<Void> {
    @Override
    public Void call() throws Exception {
        Thread.sleep(1000);
        return null;
    }
}

印刷品

Queue length 999998
After shutdown
java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@e026161 rejected from java.util.concurrent.ThreadPoolExecutor@3e472e76[Shutting down, pool size = 2, active threads = 2, queued tasks = 999998, completed tasks = 0]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2013)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:816)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1337)
    at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:132)
    at Main.main(Main.java:17)