如果消费者线程遇到异常如何停止生产者线程

Question

我有这种生产者消费者的情况，我正在使用 arrayblockingqueue。如果消费者线程遇到异常，如何停止生产者线程。我需要生产者停止等待队列为空。我引发了强制运行时异常。但程序并没有退出。生产者一直等待队列为空。有人可以帮忙吗

public class ServiceClass implements Runnable{

    private final static BlockingQueue<Integer> processQueue = new ArrayBlockingQueue<>(10);

    private static final int CONSUMER_COUNT = 1;

    private boolean isConsumerInterrupted = false;

    private boolean isConsumer = false;

    private static boolean producerIsDone = false;

    public ServiceClass(boolean consumer,boolean isConsumerInterrupted) {

        this.isConsumer = consumer;

        this.isConsumerInterrupted = isConsumerInterrupted;

    }

    public static void main(String[] args) {

        long startTime = System.nanoTime();

        ExecutorService producerPool = Executors.newFixedThreadPool(1);

        producerPool.submit(new ServiceClass(false,false)); // run method is

                                                         // called   

        // create a pool of consumer threads to parse the lines read

        ExecutorService consumerPool = Executors.newFixedThreadPool(CONSUMER_COUNT);

        for (int i = 0; i < CONSUMER_COUNT; i++) {

            consumerPool.submit(new ServiceClass(true,false)); // run method is

                                                            // called

        }

        producerPool.shutdown();

        consumerPool.shutdown();

        while (!producerPool.isTerminated() && !consumerPool.isTerminated()) {

        }

        long endTime = System.nanoTime();

        long elapsedTimeInMillis = TimeUnit.MILLISECONDS.convert((endTime - startTime), TimeUnit.NANOSECONDS);

        System.out.println("Total elapsed time: " + elapsedTimeInMillis + " ms");

    }

    @Override

    public void run() {

        if (isConsumer) {

            consume();

        } else {

            readFile(); //produce data by reading a file

        }

    }

Answer 1

您正在使用标志isConsumerInterrupted来终止生产者线程。这是错误的。消费者不会消耗队列中的元素，而生产者会持续生产直到队列已满，然后开始阻塞，直到队列未满。然后，当消费者抛出 RuntimeException 时，它会设置该标志，并且生产者线程没有机会检查该标志，因为没有消费者消耗队列中的元素，以便生产者可以摆脱等待状态。一种选择是使用 future 并在消费者抛出与设置标志相反的异常时取消它。由于它processQueue.put响应中断，因此它将成功终止生产者线程。InterruptedException如果在等待期间被中断，则会抛出An 。它看起来是这样的。private static Future<?> producerFuture = null;public static void main(String[] args) {&nbsp; &nbsp; // Remainder omitted.&nbsp; &nbsp; producerFuture = producerPool.submit(new ServiceClass(false, false));&nbsp;&nbsp; &nbsp; // ...}private void consume() {&nbsp; &nbsp; try {&nbsp; &nbsp; &nbsp; &nbsp; // ...&nbsp; &nbsp; } catch (Exception e) {&nbsp; &nbsp; &nbsp; &nbsp; producerFuture.cancel(true);&nbsp; &nbsp; }}