防止在 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();
        }
    }
}

此示例代码通过死锁它和启动代码的任何其他未来调用方来使此并发程序失败。

<小时 />

问题是：我怎样才能防止我的应用程序在高负载下产生大量线程（我希望任务排队），并且仍然防止仅由生产者污染执行器而死锁？

即使此示例明显在 100% 的时间内失败，也请考虑一个并发程序，在不幸的情况下，它只用生产者完全填充有界执行器 - 您将遇到相同的一般问题。

Answer 1

什么是死锁？Java 文档

死锁描述了两个或多个线程被阻塞的情况 永远，等待着彼此。

因此，当第一个线程持有监视器 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());
    }
}

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

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