Java 8 并行运行多个方法

Question

我有 2 个方法具有不同的返回类型，我想同时运行它们。这是我的代码：

public void method(int id) {
    final CompletableFuture<List<FooA>> fooACF = CompletableFuture.supplyAsync(() -> generateFooA(id));
    final CompletableFuture<List<FooB>> fooBCF = CompletableFuture.supplyAsync(() -> generateFooB(id));
    List<FooA> fooAs = fooACF.get();
    List<FooB> fooBs = fooBCF.get();
    //Do more processesing
}

public List<FooA> generateFooA(int id) {
    //code
}

public List<FooB> generateFooB(int id) {
    //code
}

但我不知道这两种方法是否会与上述代码并行运行，或者我是否最好说：

List<FooA> fooAs = generateFooA(id);
List<FooB> fooBs = generateFooB(id);

我如何正确使用可完成的期货才能并行运行这两种方法？

Answer 1

您的代码工作正常，使用ForkJoinPool.commonPool() 提供的线程，正如CompletableFuture.supplyAsync(Supplier<U> supplier) 的JavaDoc 所承诺的那样。您可以通过添加一些sleep() 和println() 语句以快速而简单的方式证明它。我通过使用String 而不是List<Foo> 稍微简化了您的代码：

public void method(int id) throws InterruptedException, ExecutionException {
    CompletableFuture<String> cfa = CompletableFuture.supplyAsync(() -> generateA(id));
    CompletableFuture<String> cfb = CompletableFuture.supplyAsync(() -> generateB(id));
    String fooA = cfa.get();
    String fooB = cfb.get();
    System.out.println("Final fooA " + fooA);
    System.out.println("Final fooB " + fooB);
}

public String generateA(int id) {
    System.out.println("Entering generateA " + Thread.currentThread());
    sleep(2000);
    System.out.println("Leaving generateA");
    return "A" + id;
}

public String generateB(int id) {
    System.out.println("Entering generateB " + Thread.currentThread());
    sleep(1000);
    System.out.println("Leaving generateB");
    return "B" + id;
}

private void sleep(int n) {
    try {
        Thread.sleep(n);
    } catch (InterruptedException ex) {
        // never mind
    }
}

输出是：

Entering generateFooA Thread[ForkJoinPool.commonPool-worker-1,5,main]
Entering generateFooB Thread[ForkJoinPool.commonPool-worker-2,5,main]
Leaving generateFooB
Leaving generateFooA
Final fooA A1
Final fooB B1

您可以手动观察“离开”输出行在 1 秒和 2 秒后出现。如需更多证据，您可以在输出中添加时间戳。如果您更改睡眠的相对长度，您将看到“离开”输出以不同的顺序显示。

---

如果您省略了sleep()s，那么第一个线程很可能会很快完成，以至于在第二个线程开始之前就完成了：

Entering generateA Thread[ForkJoinPool.commonPool-worker-1,5,main]
Leaving generateA
Entering generateB Thread[ForkJoinPool.commonPool-worker-1,5,main]
Leaving generateB
Final fooA A1
Final fooB B1

请注意，这一切都发生得如此之快，以至于在运行时请求第二个线程时，线程已经返回到池中。所以原来的线程被重用了。

可以想象，这也可能发生在非常短的睡眠中，尽管在我的系统上，每次运行它时 1 毫秒的睡眠就足够了。当然，sleep() 是需要时间才能完成的“真实”操作的占位符。如果您的实际操作如此便宜以至于它在另一个线程启动之前完成，那么这很好地暗示了这是一个多线程无益的场景。

---

但是如果您需要询问如何证明事情是同时发生的，我想知道您为什么首先希望它们同时发生。如果您的程序在同时或按顺序执行这些任务时没有“现实世界”可观察到的差异，那么为什么不让它按顺序运行呢？更容易推理顺序操作；有很多与并发相关的偷偷摸摸的错误。

也许您希望通过多线程来提高速度——如果是这样的话，速度的提高应该是您所衡量的，而不是事情是否实际上是并发的。请记住，对于非常多的任务，CPU 并行执行它们的速度比顺序执行更快。

Answer 2

您缺少Executor：

ExecutorService executor = Executors.newCachedThreadPool();
List<Future<?>> = Stream.<Runnable>of(() -> generateFooA(id), () -> generateFooA(id))
        .map(executor::submit)
        .collect(Collectors.toList());
for (Future<?> future : futures) {
    future.get(); // do whatever you need here
}

Runnables 在您 submit 他们时开始执行。 get() 会尽快返回。例如，如果你get() 的第一个未来是最慢的，那么所有其他get() 调用将立即返回。

Answer 3

正如我在评论中所说，请查看How to start two threads at "exactly" the same time，但这应该是你要找的东西

final CyclicBarrier gate = new CyclicBarrier(3);
public void method(int id) {
    Thread one = new Thread (()->{
        gate.await();
        List<FooA> fooAs = generateFooA(id);
    });
    Thread two = new Thread (()->{
        gate.await();
        List<FooB> fooBs = generateFooB(id);
    });
    one.start();
    two.start();
    gate.await();
    //Do more processesing
}

public List<FooA> generateFooA(int id) {
    //code
}

public List<FooB> generateFooB(int id) {
    //code
}