34-案例实现

案例实现

根据如下步骤实现本案例。

1．实现本案例中的辅助任务。创建一个名为 SeedGenerator 的类，实现 Runnable 接口。该类拥有一个 CompletableFuture 类型的属性，并在类构造器中完成初始化：

public class SeedGenerator implements Runnable {
  private CompletableFuture<Integer> resultCommunicator;
  public SeedGenerator (CompletableFuture<Integer> completable) {
    this.resultCommunicator=completable;
  }

2．实现 run() 方法。该方法会休眠当前线程5s^[5]，生成一个1和10之间的随机数字，然后调用 resultCommunicator 对象的 complete() 方法完成 CompletableFuture ：

@Override
public void run() {
  System.out.printf("SeedGenerator: Generating seed...\n");
  // Wait 5 seconds
  try {
    TimeUnit.SECONDS.sleep(5);
  } catch (InterruptedException e) {
    e.printStackTrace();
  }
    int seed=(int) Math.rint(Math.random() * 10);
    System.out.printf("SeedGenerator: Seed generated: %d\n",
                      seed);
    resultCommunicator.complete(seed);
  }

3．创建一个名为 NumberListGenerator 的类，实现泛型为 List<Long> 类型的 Supplier 接口。这意味着，由 Supplier 接口定义的 get() 方法，其返回值是一个大数列表。该类拥有一个私有的整型属性，它在类构造器中完成初始化：

public class NumberListGenerator implements Supplier<List<Long>> {
  private final int size;
  public NumberListGenerator (int size) {
    this.size=size;
  }

4．实现 get() 方法。该方法将返回有多达百万个随机数字的列表，其数量由参数 size 决定：

@Override
public List<Long> get() {
  List<Long> ret = new ArrayList<>();
  System.out.printf("%s : NumberListGenerator : Start\n",
                    Thread.currentThread().getName());
  for (int i=0; i< size*1000000; i++) {
    long number=Math.round(Math.random()*Long.MAX_VALUE);
    ret.add(number);
  }
  System.out.printf("%s : NumberListGenerator : End\n",
                    Thread.currentThread().getName());
  return ret; 
}

5．创建一个名为 NumberSelector 的类，实现由 List 和 Long 数据类型作为参数的接口 Function 。这意味着，由 Function 接口定义的 apply() 方法接收一个大数列表作为参数，并返回一个 Long 类型的数字：

public class NumberSelector implements Function<List<Long>, Long> {
  @Override
  public Long apply(List<Long> list) {
    System.out.printf("%s: Step 3: Start\n",
                      Thread.currentThread().getName());
    long max=list.stream().max(Long::compare).get();
    long min=list.stream().min(Long::compare).get();
    long result=(max+min)/2;
    System.out.printf("%s: Step 3: Result - %d\n",
                      Thread.currentThread().getName(), result);
    return result;
  }
}

6．实现本案例主程序的入口 main() 方法：

public class Main {
  public static void main(String[] args) {

7．创建一个 CompletableFuture 对象和一个 SeedGenerator 任务，并且启动线程来执行该任务：

System.out.printf("Main: Start\n");
CompletableFuture<Integer> seedFuture = new CompletableFuture<>();
Thread seedThread = new Thread(new SeedGenerator(seedFuture));
seedThread.start();

8．调用 CompletableFuture 的 get() 方法，等待 SeedGenerator 任务生成 seed ：

System.out.printf("Main: Getting the seed\n");
int seed = 0;
try {
  seed = seedFuture.get();
} catch (InterruptedException | ExecutionException e) {
  e.printStackTrace();
}
System.out.printf("Main: The seed is: %d\n", seed);

9．创建另一个 CompletableFuture 对象来控制 NumberListGenerator 任务的执行。在本案例中，用的是静态方法 supplyAsync() ：

System.out.printf("Main: Launching the list of numbers
                   generator\n");
NumberListGenerator task = new NumberListGenerator(seed);
CompletableFuture<List<Long>> startFuture = CompletableFuture
                                            .supplyAsync(task);

10．配置3个并行任务，这些任务在从之前任务中得到的数字列表上进行计算。这3个步骤只有在 NumberListGenerator 任务完成后才能开始执行，因此使用上一步生成的 CompletableFuture 对象和 thenApplyAsync() 方法来配置这些任务。前两步任务采用函数式编程实现，第三步则传入一个 NumberSelector 类：

System.out.printf("Main: Launching step 1\n");
CompletableFuture<Long> step1Future = startFuture
                                      .thenApplyAsync(list -> {
  System.out.printf("%s: Step 1: Start\n", 
                    Thread.currentThread().getName());
  long selected = 0;
  long selectedDistance = Long.MAX_VALUE;
  long distance;
  for (Long number : list) {
    distance = Math.abs(number - 1000);
    if (distance < selectedDistance) {
      selected = number;
      selectedDistance = distance;
    }
  }
  System.out.printf("%s: Step 1: Result - %d\n", 
                    Thread.currentThread().getName(), selected);
  return selected;
});
System.out.printf("Main: Launching step 2\n");
CompletableFuture<Long> step2Future = startFuture
.thenApplyAsync(list -> list.stream().max(Long::compare).get());
CompletableFuture<Void> write2Future = step2Future
                                      .thenAccept(selected -> {
  System.out.printf("%s: Step 2: Result - %d\n", 
                    Thread.currentThread().getName(), selected);
});
System.out.printf("Main: Launching step 3\n");
NumberSelector numberSelector = new NumberSelector();
CompletableFuture<Long> step3Future = startFuture
                                .thenApplyAsync(numberSelector);

11．使用 CompletableFuture 类的静态方法 allOf() ，等待这3个并行步骤的结束：

System.out.printf("Main: Waiting for the end of the three 
                   steps\n");
CompletableFuture<Void> waitFuture = CompletableFuture
                              .allOf(step1Future, write2Future,
                                    step3Future);

12．执行最后一步，向控制台输出信息：

CompletableFuture<Void> finalFuture = waitFuture
                                   .thenAcceptAsync((param) -> {
  System.out.printf("Main: The CompletableFuture example has 
                     been completed.");
});
finalFuture.join();