java - 使用 Java 中的生产者和消费者防止有界执行程序服务中可能发生的死锁情况

Question

考虑这个示例代码:(我已经大大简化了类，所以它们更容易阅读)

制作人

class RandomIntegerProducer implements Callable<Void>
{
    private final BlockingQueue<? super Integer> queue;
    private final Random random;

    /* Boilerplate constructor... */

    @Override
    public Void call()
    {
        while (!Thread.interrupted())
        {
            try {
                TimeUnit.SECONDS.sleep(1);
                queue.put(random.nextInt());
            } catch (InterruptedException e)
            {
                Thread.currentThread().interrupt();
                break;
            }
        }
        return null;
    }
}

这是一个简单、简洁的任务示例，每秒将一个随机数放入队列中，并且可以使用 Thread.interrupt() 取消。

消费者

class NumberConsumer implements Callable<Void>
{
    private final BlockingQueue<? extends Number> queue;
    private final Appendable target;

    /* Boilerplate constructor... */

    @Override
    public Void call() throws IOException
    {
        while (!Thread.interrupted())
        {
            try {
                target.append(queue.take().toString());
            } catch (InterruptedException e)
            {
                Thread.currentThread().interrupt();
                break;
            }
        }
        return null;
    }
}

消费者从队列中取出数字并将它们打印到指定的Appendable。可以通过 Thread.interrupt() 取消。

起始代码

class ProducerConsumerStarter
{
    /* Notice this is a fixed size (e.g. bounded) executor service */
    private static final ExecutorService SERVICE = Executors.newFixedThreadPool(8);

    public static List<Future<Void>> startIntegerProducerConsumer(int producers, int consumers)
    {
        List<Callable<Void>> callables = new ArrayList<>();
        BlockingQueue<Integer> commonQueue = new ArrayBlockingQueue<>(16);
        for (int i = 0; i < producers; i++)
        {
            callables.add(new RandomIntegerProducer(commonQueue, new Random()));
        }
        for (int i = 0; i < consumers; i++)
        {
            callables.add(new NumberConsumer(commonQueue, System.out));
        }
        // Submit them all (in order)
        return callables.stream().map(SERVICE::submit).collect(Collectors.toList());
    }
}

此实用程序方法将任务提交给有界执行程序服务(按顺序 - 首先是所有生产者，然后是所有消费者)

失败的客户端代码

public class FailingExaple {
    @org.junit.Test
    public void deadlockApplication() throws Exception
    {
        List<Future<Void>> futures = ProducerConsumerStarter.startIntegerProducerConsumer(10, 10);
        for (Future<Void> future : futures)
        {
            System.out.println("Getting future");
            future.get();
        }
    }
}

此示例代码通过将它和任何其他 future 的起始代码调用者死锁而使该并发程序失败。

---

问题是:我如何才能既防止我的应用程序在高负载下产生大量线程(我希望任务改为排队)，又能防止仅由生产者污染执行程序的死锁？

即使这个示例在 100% 的时间里明显失败，考虑一个并发程序，它在不幸的情况下完全用生产者填充有界执行器——你会遇到同样的一般问题。

Answer 1

什么是死锁？ Java Documentation

Deadlock describes a situation where two or more threads are blocked forever, waiting for each other.

因此，当第一个线程持有监视器 1 并尝试获取监视器 2，而第二个线程持有监视器 2 并尝试获取监视器 1 时，就会发生死锁。
您的代码中没有死锁，因为没有 two or more threads .. waiting for each other .有生产者在等待队列中的空间，没有消费者，因为由于执行者的线程数，他们没有被安排。

此外，“失败的客户端代码” 将始终阻塞线程，即使是 startIntegerProducerConsumer(1,1)

public class FailingExaple {
    @org.junit.Test
    public void deadlockApplication() throws Exception
    {
        List<Future<Void>> futures = ProducerConsumerStarter.startIntegerProducerConsumer(10, 10);
        for (Future<Void> future : futures)
        {
            System.out.println("Getting future");
            future.get();
        }
    }
}

因为您的生产者和消费者一直在运行，直到发生明确的中断，这在 deadlockApplication() 中不会发生。 .

你的代码应该是这样的

for (Future<Void> future : futures)
{
    if (future.isDone()) {
        try {
            System.out.println("Getting future");
            future.get();
        } catch (CancellationException ce) {

        } catch (ExecutionException ee) {

        }
    } else {
        System.out.println("The future is not done, cancelling it");
        if (future.cancel(true)) {
            System.out.println("task was cancelled");
        } else {
            //handle case when FutureTask#cancel(boolean mayInterruptIfRunning) wasn't cancelled
        }
    }
}

此循环将获取已完成任务的结果并取消未完成。

@vanOekel 是的，最好有两个线程池，一个给消费者，一个给生产者。
像这样

class ProducerConsumerStarter
{
    private static final ExecutorService CONSUMERS = Executors.newFixedThreadPool(8); 
    private static final ExecutorService PRODUCERS = Executors.newFixedThreadPool(8); 

    public static List<Future<Void>> startIntegerProducerConsumer(int producers, int consumers) {
        ...
    }
}

和startIntegerProducerConsumer(int, int)相应地提交消费者和生产者。
但在这种情况下，新的任务将排队等待之前提交的生产者和消费者完成(如果这些任务不被中断，则不会发生)。

您还可以进一步优化生产者的代码。首先更改代码

class RandomIntegerProducer implements Runnable
{
    private final BlockingQueue<? super Integer> queue;
    private final Random random;
    ...
    @Override
    public void run()
    {
        queue.offer(random.nextInt());
    }
}

然后开始将生产者提交到ScheduledExecutorService使用 scheduleWithFixedDelay(producer, 1, 1, TimeUnit.SECONDS) .此更改将有助于保持生产者运行而不会相互阻塞。但它也会稍微改变应用程序的语义。
你可以保留ScheduledExecutorService (对于生产者)初始化为类变量。唯一的不便是您必须更改 startIntegerProducerConsumer(int producers, int consumers) 的返回类型List<Future<?>> 的方法但实际上 ScheduledFutures<?>返回scheduleWithFixedDelay(..)仍然是 Future<Void> 类型.在使用新生成的数字期间，如果可能的话，您可以对消费者执行相同的操作，最大延迟等于 delay。 (传递给 scheduleWithFixedDelay() )适合你。

希望我的回答对您有所帮助。