java - 不断向 ExecutorService 提交 Runnable 任务，直到工作完成，得到 java.util.concurrent.RejectedExecutionException

Question

我的多线程类应该对 ClassA 类的多个对象执行三个操作 - 操作 1、操作 2 和 操作 3，其中每种类型的操作都依赖于先前的操作。为此，我尝试使用多个 BlockingQueue 和 ExecutorService 来实现生产者-消费者模式。

final ExecutorService executor = ForkJoinPool.commonPool();
final BlockingQueue<ClassA> operationOneQueue = new ArrayBlockingQueue<>(NO_OF_CLASS_A_OBJECTS);
final BlockingQueue<ClassA> operationTwoQueue = new ArrayBlockingQueue<>(NO_OF_CLASS_A_OBJECTS);
final BlockingQueue<ClassA> operationThreeQueue = new ArrayBlockingQueue<>(NO_OF_CLASS_A_OBJECTS);
final BlockingQueue<ClassA> resultQueue = new ArrayBlockingQueue<>(NO_OF_CLASS_A_OBJECTS);

操作的实现如下:

void doOperationOne() throws InterruptedException {
    ClassA objectA = operationOneQueue.take();
    objectA.operationOne();
    operationTwoQueue.put(objectA);
}

其中每种类型的操作都有其自己相应的方法，以及其“自己的”入队列和出队列。每个操作方法都会调用 ClassA 对象上适当的方法。 doOperationThree 方法将 ClassA 对象放入 resultQueue 中，这意味着它们已被完全处理。

首先，我将要操作的所有 ClassA 对象填充到 operationOneQueue 中。然后，我尝试将可执行任务分配给 ExecutorService，如下所示:

    while (resultQueue.size() < NO_OF_CLASS_A_OBJECTS) {
        executor.execute(() -> {
            try {
                doOperationOne();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        executor.execute(() -> {
            try {
                doOperationTwo();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        executor.execute(() -> {
            try {
                doOperationThree();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
    }
    executor.shutdown();

运行我的程序时，我收到一个java.util.concurrent.RejectedExecutionException。

    Operation1: ClassA object 0
    Operation2: ClassA object 0
    Operation1: ClassA object 1
    Operation3: ClassA object 0
    ....
    Operation1: ClassA object 46
    Operation2: ClassA object 45
    Operation3: ClassA object 45
    Exception in thread "main" java.util.concurrent.RejectedExecutionException: Queue capacity exceeded
at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.growArray(ForkJoinPool.java:912)
at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.lockedPush(ForkJoinPool.java:867)
at java.base/java.util.concurrent.ForkJoinPool.externalPush(ForkJoinPool.java:1911)
at java.base/java.util.concurrent.ForkJoinPool.externalSubmit(ForkJoinPool.java:1930)
at java.base/java.util.concurrent.ForkJoinPool.execute(ForkJoinPool.java:2462)
at concurrent.operations.Program1.main(Program1.java:96)

我做错了什么？如何在线程池不过度饱和的情况下实现这一目标？

编辑:全面披露——这是有一些要求的作业。 1. 我必须使用 ForkJoinPool.commonPool() 并且不能自己设置线程数，2. 我必须使用消费者-生产者模式，3. 我不能修改 ClassA。

Answer 1

我真的很喜欢做并发的事情，所以我尝试编写它。我确实使用了 CompletableFuture，它 a) 默认情况下在 ForkJoinPool.commonPool 中运行，b) 使实际处理变得非常简单:

while (true) {
    final ClassA nextOperation = queue.take();
    CompletableFuture.runAsync(nextOperation::operationOne)
        .thenRun(nextOperation::operationTwo)
        .thenRun(nextOperation::operationThree)
        .thenRun(() -> resultQueue.add(nextOperation));
}

这将从队列中获取ClassA对象并同时执行它们的所有操作，但按顺序。

您确实遗漏了任务的来源以及是否需要消费者终止。通常您不想这样做，但这确实会让事情变得更复杂。

private static final int COUNT = 10;
private static final Random RANDOM = new Random();

public static void main(String[] args) throws ExecutionException, InterruptedException {
    BlockingQueue<ClassA> runnables = new ArrayBlockingQueue<>(COUNT);
    BlockingQueue<ClassA> finished = new ArrayBlockingQueue<>(COUNT);

    // start producer
    ExecutorService createTaskExecutor = Executors.newSingleThreadExecutor();
    createTaskExecutor.submit(() -> fillQueue(runnables));
    // wait for all consumer tasks to finish
    while (finished.size() != COUNT) {
        try {
            // we need to poll instead of waiting forever
            // because the last tasks might still be running
            // while there are no others to add anymore
            // so we need to check again if all have finished in the meantime
            final ClassA nextOperation = runnables.poll(2, TimeUnit.SECONDS);
            if (nextOperation != null) {
                CompletableFuture.runAsync(nextOperation::operationOne)
                        .thenRun(nextOperation::operationTwo)
                        .thenRun(nextOperation::operationThree)
                        .thenRun(() -> finished.add(nextOperation));
            }
        } catch (InterruptedException e) {
            System.err.println("exception while retrieving next operation");
            // we will actually need to terminate now, or probably never will
            throw e;
        }
    }
    System.out.printf("finished tasks (%d):%n", finished.size());
    for (ClassA classA : finished) {
        System.out.printf("finished task %d%n", classA.designator);
    }
    createTaskExecutor.shutdown();
}

private static void fillQueue(BlockingQueue<ClassA> runnables) {
    // start thread filling the queue at random
    for (int i = 0; i < COUNT; i++) {
        runnables.add(new ClassA(i));
        try {
            Thread.sleep(RANDOM.nextInt(1_000));
        } catch (InterruptedException e) {
            System.err.println("failed to add runnable");
        }
    }
}

由于您没有提供ClassA，所以我使用了这个。它包含一个标识符，以便您可以跟踪哪个程序在什么时间运行。

class ClassA {
    private static final Random RANDOM = new Random();
    public final int designator;

    public ClassA(int i) {
        designator = i;
    }

    public void operationOne() {
        System.out.printf("%d: operation 1%n", designator);
        sleep();
    }

    public void operationTwo() {
        System.out.printf("%d: operation 2%n", designator);
        sleep();
    }

    public void operationThree() {
        System.out.printf("%d: operation 3%n", designator);
        sleep();
    }

    private static void sleep() {
        try {
            Thread.sleep(RANDOM.nextInt(5_000));
        } catch (InterruptedException e) {
            System.err.println("interrupted while executing task");
        }
    }
}