Пул потоков Java с ограниченной очередью

Я использую класс java.util.concurrent Executors, чтобы создать фиксированный пул потоков для запуска обработчиков запросов для веб-сервера:

static ExecutorService  newFixedThreadPool(int nThreads)

и описание:

Создает пул потоков, который повторно использует фиксированный набор потоков, работающих с общей неограниченной очередью.

Однако я ищу реализацию пула потоков, которая будет делать то же самое, за исключением очереди ограниченного. Есть ли такая реализация? Или мне нужно реализовать свою собственную оболочку для фиксированного пула потоков?

Ответ 1

Что вы хотите сделать, это новый собственный ExecutorService, возможно, используя ThreadPoolExecutor. ThreadPoolExecutor имеет конструктор, который принимает BlockingQueue и получает ограниченную очередь, которую вы используете, например ArrayBlockingQueue, правильно построенный для ограничения. Вы также можете включить RejectedExecutionHandler, чтобы определить, что делать, когда ваша очередь заполнена, или привязаться к ссылке на блокирующую очередь и используйте методы предложения.

Вот мини-пример:

BlockingQueue<Runnable> linkedBlockingDeque = new LinkedBlockingDeque<Runnable>(
    100);
ExecutorService executorService = new ThreadPoolExecutor(1, 10, 30,
    TimeUnit.SECONDS, linkedBlockingDeque,
    new ThreadPoolExecutor.CallerRunsPolicy());

Ответ 2

Создайте ThreadPoolexecutor и передайте в него подходящую реализацию BlockingQueue. например, вы можете передать ArrayBlockingQueue в конструкторе ThreadPoolExecutor, чтобы получить желаемый эффект.

Ответ 3

При создании ThreadPoolExecutor вы можете дать ему ограниченный BlockingQueue и RejectedExecutionHandler, чтобы вы могли контролировать, что происходит, когда предел достигнут. Поведение по умолчанию - это выбросить исключение RejectedExecutionException.

Вы также можете определить свой собственный поток factory, чтобы управлять именами потоков и создавать их потоки демона.

Ответ 4

Я решил это с помощью Семафора, который я использую для дросселирования заданий, передаваемых в ExecutorService.

Например:

int threadCount = 10;
ExecutorService consumerPool = Executors.newFixedThreadPool(threadCount);

// set the permit count greater than thread count so that we 
// build up a limited buffer of waiting consumers
Semaphore semaphore = new Semaphore(threadCount * 100); 

for (int i = 0; i < 1000000; ++i) {
    semaphore.acquire(); // this might block waiting for a permit 
    Runnable consumer = () -> {
       try {
          doSomeWork(i);
       } finally {
          semaphore.release(); // release a permit 
       }
    };
    consumerPool.submit(consumer);
}