【问题标题】：Java 8 Completable Future - parallel executionJava 8 Completable Future - 并行执行
【发布时间】：2019-03-23 01:15:03
【问题描述】：

我正在测试CompletableFuture 的工作原理。我对如何并行执行任务感兴趣：

try {
          CompletableFuture one = CompletableFuture.runAsync(() -> {
          throw new RuntimeException("error");
          });
          CompletableFuture two = CompletableFuture.runAsync(() -> System.out.println("2"));
          CompletableFuture three = CompletableFuture.runAsync(() -> System.out.println("3"));
          CompletableFuture all = CompletableFuture.allOf(one, two, three);
          all.get();
} catch (InterruptedException e) {
           System.out.println(e);
} catch (ExecutionException e) {
           System.out.println(e);
}

在这种情况下，它们将被全部执行。

1。当其中一个线程出现异常时，是否可以中断所有正在运行的线程？

2。当这段代码在一个可以从不同线程调用的类方法中时，它是线程安全的吗？

【问题讨论】：

标签： java multithreading java-8 completable-future

【解决方案1】：

1.当其中一个线程出现异常时，是否可以中断所有正在运行的线程？

是的，这是可能的。所有线程都应该可以访问公共对象，该对象的状态可以被其他线程更改和读取。例如，它可以是AtomicInteger。见下例：

import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

public class Dates {

    public static void main(String[] args) throws Exception {
        try {
            AtomicInteger excCounter = new AtomicInteger(0);
            CompletableFuture one = CompletableFuture.runAsync(new ExcRunnable(excCounter));
            CompletableFuture two = CompletableFuture.runAsync(new PrintRunnable("2", excCounter));
            CompletableFuture three = CompletableFuture.runAsync(new PrintRunnable("3", excCounter));
            CompletableFuture all = CompletableFuture.allOf(one, two, three);
            all.get();
        } catch (InterruptedException | ExecutionException e) {
            System.out.println(e);
        }
    }
}

class ExcRunnable implements Runnable {
    private final AtomicInteger excCounter;

    public ExcRunnable(AtomicInteger excCounter) {
        this.excCounter = excCounter;
    }

    @Override
    public void run() {
        Random random = new Random();
        int millis = (int) (random.nextDouble() * 5000);
        System.out.println("Wait " + millis);
        Threads.sleep(450);

        // Inform another threads that exc occurred
        excCounter.incrementAndGet();

        throw new RuntimeException("error");
    }
}

class PrintRunnable implements Runnable {

    private final String name;
    private final AtomicInteger excCounter;

    public PrintRunnable(String name, AtomicInteger excCounter) {
        this.name = name;
        this.excCounter = excCounter;
    }

    @Override
    public void run() {
        int counter = 10;
        while (counter-- > 0 && excCounter.get() == 0) {
            System.out.println(name);
            Threads.sleep(450);
        }
    }
}

class Threads {
    static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

我们有 3 个任务：两个打印它的名字，一个在一段时间后抛出异常。在抛出异常之前，计数器会递增以通知其他任务其中一个失败并且它们应该完成执行。打印作业正在检查这个计数器，如果条件不满足，他们会完成它的工作。当您评论 excCounter.incrementAndGet(); 行时，其他任务在不知道其中一个引发异常的情况下完成了他们的工作。

当此代码位于可以从不同线程调用的类方法中时，它是线程安全的吗？

看一下thread safety的定义。例如，假设打印任务每打印一行都会增加公共计数器。如果计数器是原始的int，则它不是线程安全的，因为可以替换计数器值。但是如果你使用AtomicInteger 是线程安全的，因为AtomicInteger 是线程安全的。

【讨论】：

很好，它也适用于 ExecutorService，谢谢！

【解决方案2】：

1

如果您的任何异步任务抛出异常， all.get() 将抛出异常。这意味着，您可以在 catch 子句中取消所有 CF。但是，您的异步任务需要对中断友好，即定期检查中断标志或处理 InterruptedException 并提前返回。

应始终使用中断机制处理任务取消

2

您提到的所有引用变量都是本地的，因此无需担心线程安全。局部变量始终是线程安全的。

【讨论】：