在 Java 中使用 ExecutorService 执行第一个可运行对象时是否可以阻止其他可运行对象？答案

【问题标题】：Is it possible to block other runnables while executing the first one, using ExecutorService in Java?在 Java 中使用 ExecutorService 执行第一个可运行对象时是否可以阻止其他可运行对象？
【发布时间】：2020-09-03 05:31:26
【问题描述】：

我正在尝试使用ExecutorService 在多个线程中处理相对较大的Stream 或List。该方法看起来像这样。

public void initMigration() {
    ExecutorService executorService = Executors.newCachedThreadPool();
    try (Stream<List<Record4<Integer, Integer, String, byte[]>>> streamOfLists = getStreamOfLists()) {            
        streamOfLists.forEach(record4List -> {
            Runnable runnable = () -> {
                try {
                    final List<Attachment> attachments = RecordProcessor.prepareAttachmentsToPost(record4List);
                    LOGGER.info("Invoking POST with payload {}", attachments);
                    Collection<UploadLink> uploadLinks = restClient.postAttachments(attachments);
                    restClient.processUploadLinksAndUpload(RecordProcessor.recordsIntoPojo(record4List), uploadLinks);
                } catch (ExceptionA | ExceptionB e) {
                    e.printStackTrace();
                }
            };
            executorService.submit(runnable);
        });
    }
    LOGGER.info("Shutting down the ExecutorService");
    executorService.shutdown();
}

基本上，我在这里要做的是，对于Stream 中的每个List，正在创建一个Runnable 并将其提交给ExecutorService。它似乎工作正常。但是，我现在真正想做的是看看是否有任何方法可以让ExecutorService 运行从Stream 中的第一个List 获得的第一个Runnable，同时阻止其他Runnables 直到它的执行，然后继续运行其他Runnables（并行）。真的可以在这方面使用一些帮助。

【问题讨论】：

标签： java executorservice java-threads

【解决方案1】：

您可以先获取 Runnable，执行它，然后才能提交其他 Runnable。

    try (Stream<List<Record4<Integer, Integer, String, byte[]>>> streamOfLists = getStreamOfLists()) {
        Iterator<List<Record4<Integer, Integer, String, byte[]>>> it = streamOfLists.iterator();
        if (it.hasNext()) {
            List<Record4<Integer, Integer, String, byte[]>> list = it.next();
            Runnable runnable = new MyRunnable(record4List);
            runnable.run();
        }
        while (it.hasNext()) {
            List<Record4<Integer, Integer, String, byte[]>> list = it.next();
            Runnable runnable = new MyRunnable(record4List);
            executorService.submit(runnable);
        }
    }

在哪里

class MyRunnable implements Runnable {
    Record4<Integer, Integer, String, byte[]> record4List;

    MyRunnable(Record4<Integer, Integer, String, byte[]> record4List) {
        this.record4List = record4List;
    }

    @Override
    public void run() {
        try {
            final List<Attachment> attachments = RecordProcessor.prepareAttachmentsToPost(record4List);
            LOGGER.info("Invoking POST with payload {}", attachments);
            Collection<UploadLink> uploadLinks = restClient.postAttachments(attachments);
            restClient.processUploadLinksAndUpload(RecordProcessor.recordsIntoPojo(record4List), uploadLinks);
        } catch (ExceptionA | ExceptionB e) {
            e.printStackTrace();
        }
    }
}

【讨论】：

非常感谢您的时间和精力。这正是我所需要的。

【解决方案2】：

@Alexei's approach 是（IMO）解决此问题的正确方法。不要阻止可运行文件。而是在满足运行它们的先决条件时提交它们。

拥有一个可运行块的问题在于，您很可能会阻塞执行器的线程池，而这些任务被阻塞等待另一个任务完成。实际上，如果线程池是有界的，您甚至可能会遇到所有线程都处于这种状态并且执行器无法启动将解除所有线程阻塞的任务的情况。结果：死锁！

如果您仍然想要阻止可运行对象（尽管有上述情况），那么您可以使用CountDownLatch 来实现它。

在实例化Runnables 之前，创建一个初始计数器为1 的CountDownLatch 实例。此实例必须由所有Runnables 共享。
编码一个Runnable，以便它获取List，处理它，然后调用latch.count()；
编写第二个Runnable 以调用latch.await()，然后获取并处理List。
使用第一个 Runnable 提交一个任务，其余的使用第二个。

【讨论】：