java - 如何实现阻塞ThreadPoolExecutor

我需要实现阻塞ThreadPoolExecutor。

这是我们企业应用程序中非常关键的要求。

如果ThreadPoolExecutor.submit()或ThreadPoolExecutor.execute()方法阻塞，直到某个线程被释放以接管新任务时，它就会执行类似的操作。

但是在当前实现中，如果所有池化线程都变得繁忙，则ThreadPoolExecutor.submit()和ThreadPoolExecutor.execute()方法将引发RejectedExecutionException异常。

例如，以下代码抛出RejectedExecutionException:

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class BlockingTPE {
    public static void main(String[] args) {
        ArrayBlockingQueue queue = new ArrayBlockingQueue(3);
        ThreadPoolExecutor tpExe = new ThreadPoolExecutor(1, 3, 30, TimeUnit.SECONDS, queue);
        int numJobs = 50;
        for (int i = 1; i <= numJobs; i++) {
            try {
                tpExe.submit(new WorkerThread(i));
                System.out.println("Added#" + (i));
            } catch (RejectedExecutionException e) {
                e.printStackTrace();
            }
        }
    }
}

class WorkerThread implements Runnable {
    int jobId;

    public WorkerThread(int jobId) {
        this.jobId = jobId;
    }

    public void run() {
        try {
            Thread.sleep(1000);
        }
        catch (Exception excep) {
        }
    }
}

最佳答案

如ThreadPoolExecutor的javadoc所述:



被拒绝的执行程序处理程序是AbortPolicy的实例，如果队列不接受其他任务，则会调用该实例。从javadoc开始的行为:



因此，阻塞队列对您没有任何影响。我以这种方式更改了您的代码，它的运行没有任何问题:

public static void main(String[] args) {
    ThreadPoolExecutor executor = new ThreadPoolExecutor(
            1, 3, 30, TimeUnit.SECONDS, new ArrayBlockingQueue(3));
    try {
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        int numJobs = 50;
        for (int i = 1; i <= numJobs; i++) {
            try {
                executor.submit(new WorkerThread(i));
                System.out.println("Added#" + (i));
            } catch (RejectedExecutionException e) {
                e.printStackTrace();
            }
        }
    } finally {
        executor.shutdown();
    }
}

您必须做出的决定是:

使用未绑定(bind)的队列来支持所有延迟的任务。例如LinkedBlockingQueue。

使用绑定(bind)队列，让当前线程执行不适合整个队列的任务。例如，查看我在答案中发布的代码。

如果边界队列已满，则

放弃任务。例如，使用ThreadPoolExecutor.DiscardPolicy作为拒绝的执行处理程序。

关于java - 如何实现阻塞ThreadPoolExecutor，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/43109547/