具有有界队列的Java线程池

I'm using java.util.concurrent's Executors class to create a fixed thread pool for running request handlers for a web server:

我正在使用java.util.concurrent的Executors类来创建一个固定的线程池,用于运行Web服务器的请求处理程序:

static ExecutorService  newFixedThreadPool(int nThreads)

and the description is:

而描述是:

Creates a thread pool that reuses a fixed set of threads operating off a shared unbounded queue.

创建一个线程池,该线程池重用一组在共享*队列中运行的固定线程。

However, I am looking for thread pool implementation which will do the exact same thing, except with a bounded queue. Is there such an implementation? Or do I need to implement my own wrapper for the fixed thread pool?

但是,我正在寻找线程池实现,它将执行完全相同的操作,除了有界队列。有没有这样的实施?或者我是否需要为固定线程池实现自己的包装器?

4 个解决方案

#1

What you want to do is new your own ExecutorService, probably using ThreadPoolExecutor. ThreadPoolExecutor has a constructor which takes a BlockingQueue and to get a bounded queue you use for example ArrayBlockingQueue properly constructed for bounding. You can also include a RejectedExecutionHandler in order to determine what to do when your queue is full, or hang on to a reference to the blocking queue and use the offer methods.

你想要做的是新建自己的ExecutorService,可能使用ThreadPoolExecutor。 ThreadPoolExecutor有一个构造函数,它接受一个BlockingQueue并获得一个有界的队列,例如正确构造用于边界的ArrayBlockingQueue。您还可以包含RejectedExecutionHandler,以确定队列已满时要执行的操作,或者挂起对阻塞队列的引用并使用offer方法。

Here's a mini example:

这是一个小例子:

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

#2

Create a ThreadPoolexecutor and pass suitable BlockingQueue implementation in it. for e.g. you can pass in a ArrayBlockingQueue in the ThreadPoolExecutor constructor to get the desired effect.

创建一个ThreadPoolexecutor并在其中传递合适的BlockingQueue实现。例如您可以在ThreadPoolExecutor构造函数中传入ArrayBlockingQueue以获得所需的效果。

#3

When you create a ThreadPoolExecutor you can give it a bounded BlockingQueue and a RejectedExecutionHandler so you can control what happens when the limit is reached. The default behaviour is to throw a RejectedExecutionException.

当您创建ThreadPoolExecutor时,您可以为其提供有界的BlockingQueue和RejectedExecutionHandler,以便您可以控制达到限制时发生的情况。默认行为是抛出RejectedExecutionException。

You can also define you own thread factory to control the thread names and make them daemon threads.

您还可以定义自己的线程工厂来控制线程名称并使它们成为守护程序线程。

#4

I've solved this with a Semaphore which I use to throttle tasks being submitted to the ExecutorService.

我用Semaphore解决了这个问题,我用它来限制提交给ExecutorService的任务。

Eg:

int threadCount = 10;
ExecutorService producerPool = Executors.newSingleThreadedExecutor();
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); 

Runnable producer = () -> {
    for (int i = 0; i < 1000000; ++i) {
        semaphore.acquire();
        Runnable consumer = () -> {
           try {
              doSomeWork(i);
           } finally {
              semaphore.release();
           }
        };
        consumerPool.submit(consumer);
    }
}

Future<Void> future = producerPool.submit(producer);

// all consumers added to the pool when this returns
future.get();

producerPool.shutdown();
producerPool.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
consumerPool.shutdown();

// all consumers finished when this returns
consumerPool.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);

#1