java - 如何使用线程池为生产者-消费者解决方案创建动态消费者任务？

Question

我正在尝试实现生产者-消费者解决方案。

但我不想使用固定数量的消费者线程。相反，如果我的事件队列已满，我想创建一个新的消费者线程。

我创建了一个 ExecutorService，但由于我只有一个 EventConsumerTask 实例，因此它仅为该任务创建一个线程。

LinkedBlockingQueue<String> eventQueue = new LinkedBlockingQueue<>(50);

ExecutorService es = new ThreadPoolExecutor(5, 20, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());

es.execute(new EventConsumerTask(eventQueue));

这是我的 EventConsumerTask；

public class EventConsumerTask implements Runnable{

    private LinkedBlockingQueue<String> eventQueue;

    public EventConsumerTask(LinkedBlockingQueue<String> eventQueue){
        this.eventQueue = eventQueue;
    }

    @Override
    public void run() {
        while(true) {
            try {
                String event = eventQueue.take();
                System.out.println(event);
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

Answer 1

如果我理解正确的话，您想在队列已满时生成另一个 EventConsumerTask 吗？我想，会有一个方法可以填充队列，这样你就可以在那里触发它:

    public synchronized void add(String item){
    if(eventQueue.remainingCapacity() == 0)
    {
        es.execute(new EventConsumerTask(eventQueue)); 
    }
    waitUntilQueueHasCapacityAgain();
    eventQueue.add(item);
}

如果您存储 EventConsumerTask，例如在列表中，如果队列变空，您可以再次缩小执行程序/线程的规模。

顺便说一句。

• 您不需要 Thread.sleep()，来自 Java documentation :queue.take() 检索并删除此队列的头部，如有必要，则等待直到有元素可用。

• A ConcurrentLinkedQueue将在不锁定的情况下执行类似的相同工作，您需要在没有可用项目时才 sleep ，机器人，而不是在每个项目之后。