Java的CompletableFuture和线程

Question

我想用CompletableFuture Java8-9启动线程，使用异步模式，下面是我的类和线程：

我有3个线程。我的类只包含一个方法myMethod()

Class_1 class_1 = new Class_1();

Class_2 class_2 = new Class_2(); 

Class_3 class_3 = new Class_3();

设置我的Runnable，如下所示：

Runnable runnableClass_1 = new Runnable(){
    public void run(){
        class_1.myMethod();
        try { Thread.sleep(0); } catch (InterruptedException e) { e.printStackTrace(); }
    }
};

Runnable runnableClass_2 = new Runnable(){
    public void run(){
        class_2.myMethod();
        try { Thread.sleep(0); } catch (InterruptedException e) { e.printStackTrace(); }
    }
};

Runnable runnableClass_3 = new Runnable(){
    public void run(){
        class_3.myMethod();
        try { Thread.sleep(0); } catch (InterruptedException e) { e.printStackTrace(); }
    }
};  

创建线程：

Thread t_1 = new Thread( runnableClass_1 );

Thread t_2 = new Thread( runnableClass_2 );

Thread t_3 = new Thread( runnableClass_3 );

最后，我的问题是如何使用CompletableFuture异步模式启动这三个线程。

Answer 1

下面是如何实现相同的方法：

List<String> results = new ArrayList<String>();

        CompletableFuture<Void> run1 = CompletableFuture.runAsync(() -> {
                pauseSeconds(2);
                results.add("first task");
            }, service);

        CompletableFuture<Void> run2 = CompletableFuture.runAsync(() -> {
                pauseSeconds(3);
                results.add("second task");
            }, service);

        CompletableFuture<Void> finisher = run1.runAfterBothAsync(run2,
                                            () -> results.add(results.get(0)+ "&"+results.get(1)),service);
         pauseSeconds(4);
         System.out.println("finisher.isDone() = "+finisher.isDone());
         System.out.println("results.get(2) = "+results.get(2));
//       assert(finisher.isDone(), is(true));
//       assertThat(results.get(2),is("first task&second task"));
        }

public static void pauseSeconds(int num){
        try {
            Thread.sleep(num);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

Answer 2

你如何建立(和可能组合)你的未来取决于你的用例:这些未来相互依赖吗？您需要按顺序执行它们，还是可以并行运行它们？您是否关心所有三个结果，或者您只需要先完成哪个未来？

根据您的回答，您可以使用flatMap/bind组合器(它们对CompletableFuture有不同的名称，但您可以解决这个问题)来顺序地链接您的未来，或者您可以从当前线程中生成所有的未来(让它们并行运行)，然后等待所有的未来都完成。您还可以指定一个特定的线程池，以便只使用默认值(ForkJoinPool)来CompletableFuture工厂方法。

所有这一切都可以通过vavr提供的Future的单行版本非常简洁地完成。但是如果你看看它的documentation，你也可以提出一个使用CompletableFuture的解决方案。

更新/请求示例

以下示例基本上取自Java 8 in Action github repository，其中提供的期货并行运行，并将所有期货的结果累加到一个集合中。您要做的是，将List<Future<T>>转换为Future<List<T>>。

    final long startTime = System.currentTimeMillis();
    final CompletableFuture<String> foo = CompletableFuture.supplyAsync(() -> {
        final long timeout = 500;
        try {
            Thread.sleep(timeout);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(String.format("foo on %s sleeping for %s", Thread.currentThread(), timeout));
        return "foo";
    });
    final CompletableFuture<String> bar = CompletableFuture.supplyAsync(() -> {
        final long timeout = 100;
        try {
            Thread.sleep(timeout);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(String.format("bar on %s sleeping for %s", Thread.currentThread(), timeout));
        return "bar";
    });
    final CompletableFuture<String> baz = CompletableFuture.supplyAsync(() -> {
        final long timeout = 1000;
        try {
            Thread.sleep(timeout);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(String.format("baz on %s sleeping for %s", Thread.currentThread(), timeout));
        return "baz";
    });

    CompletableFuture
            .supplyAsync(() -> Stream.of(foo, bar, baz).map(future -> future.join()).collect(Collectors.toList()))
            .thenAccept(done -> System.out.println(String.format("Done with all futures %s", done)))
            .thenRun(() -> System.out.println(String.format("Running all futures in parallel took %s millis", System.currentTimeMillis() - startTime)));

输出应该是这样的：

bar on Thread[ForkJoinPool.commonPool-worker-2,5,main] sleeping for 100
foo on Thread[ForkJoinPool.commonPool-worker-9,5,main] sleeping for 500
baz on Thread[ForkJoinPool.commonPool-worker-11,5,main] sleeping for 1000
Done with all futures [foo, bar, baz]
Running all futures in parallel took 1007 millis