高并发编程-异步-JDK8-CompletableFuture(三)

七.任务交互

　　任务交互就是指两个线程获取任务速度的比较，按照规则进行下一步的处理。

　1.applyToEither

　　两个线程任务进行比较,先获得执行结果的，执行下一步的操作,有返回值

public <U> CompletionStage<U> applyToEither(CompletionStage<? extends T> other,Function<? super T, U> fn);public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn);public <U> CompletionStage<U> applyToEitherAsync(CompletionStage<? extends T> other,Function<? super T, U> fn,Executor executor);

 代码实例

public static void main(String[] args) throws InterruptedException {
applyToEitherDemo();
countDownLatch.await();


}
@SneakyThrows
public static void  applyToEitherDemo(){
CompletableFuture<Integer> futrue1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
@SneakyThrows
@Override
public Integer get() {
int number = new Random().nextInt(10);
System.out.println(Thread.currentThread().getName() + "第一个任务线程获得的随机数：" + number);
Thread.sleep(5000);
retu number;
}
});


CompletableFuture<Integer> futrue2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {
@SneakyThrows
@Override
public Integer get() {
int number = new Random().nextInt(10);
System.out.println(Thread.currentThread().getName() + "第二个任务线程获得的随机数：" + number);
Thread.sleep(2000);
retu number;
}
});

CompletableFuture<Integer> futrue3 = futrue1.applyToEither(futrue2, new Function<Integer, Integer>() {


@Override
public Integer apply(Integer num) {
System.out.println("查询最快的数字为" + num);

retu num * 2;
}
});

System.out.println(futrue3.get());

}

 　　2.acceptEither

　　　　两个线程任务相比较，先获得执行结果的，就对该结果进行下一步的消费操作,无返回值

public CompletionStage<Void> acceptEither(CompletionStage<? extends T> other,Consumer<? super T> action);public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action);public CompletionStage<Void> acceptEitherAsync(CompletionStage<? extends T> other,Consumer<? super T> action,Executor executor);

　　使用实例

 public static void main(String[] args) throws InterruptedException {acceptEitherDemo();countDownLatch.await();}@SneakyThrowspublic static void  acceptEitherDemo(){CompletableFuture<Integer> futrue1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@SneakyThrows@Overridepublic Integer get() {int sum = new Random().nextInt(20);Thread.sleep(5000);System.out.println(Thread.currentThread().getName() + "第一阶段" + sum);retu sum;}});CompletableFuture<Integer> futrue2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@SneakyThrows@Overridepublic Integer get() {int sum = new Random().nextInt(20);Thread.sleep(7000);System.out.println(Thread.currentThread().getName() + "第二阶段" + sum);retu sum;}});CompletableFuture<Void> result = futrue1.acceptEither(futrue2, new Consumer<Integer>() {@Overridepublic void accept(Integer sum) {System.out.println("先结束的数值为" + sum);}});System.out.println(result.get());}

　　3.runAfterEither

　　　　两个线程相比较，有任何一个线程执行完毕就执行下一步的操作，不考虑线程的返回结果。

public CompletionStage<Void> runAfterEither(CompletionStage<?> other,Runnable action);public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action);public CompletionStage<Void> runAfterEitherAsync(CompletionStage<?> other,Ru nnable action,Executor executor);

　　使用实例

public static void runAfterEitherDemo(){CompletableFuture<Integer> futrue1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@SneakyThrows@Overridepublic Integer get() {int sum = new Random().nextInt(10);Thread.sleep(1000);System.out.println(Thread.currentThread().getName() + "获取的随机数为" + sum);retu sum;}});CompletableFuture<Integer> futrue2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@SneakyThrows@Overridepublic Integer get() {int sum = new Random().nextInt(10);Thread.sleep(3000);System.out.println(Thread.currentThread().getName() + "获取的随机数为" + sum);retu sum;}});futrue2.runAfterEither(futrue1, new Runnable() {@Overridepublic void run() {System.out.println("已经有线程执行完毕");

　　　　　　　　　　countDownLatch.countDown();
}

} }); }

　　4.anyOf

　　anyOf 方法的参数是多个给定的 CompletableFuture，当其中的任何一个完成时，方法返回这个 CompletableFuture。

public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

　　使用实例

　　

@SneakyThrowspublic static void anyOfDemo(){ExecutorService executorService = Executors.newFixedThreadPool(3);CompletableFuture<Integer> futrue1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@SneakyThrows@Overridepublic Integer get() {int sum = new Random().nextInt(30);Thread.sleep(3000);System.out.println(Thread.currentThread().getName() + "开始执行：" + sum);retu sum;}},executorService);CompletableFuture<Integer> futrue2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@SneakyThrows@Overridepublic Integer get() {int sum = new Random().nextInt(30);Thread.sleep(5000);System.out.println(Thread.currentThread().getName() + "开始执行：" + sum);retu sum;}},executorService);CompletableFuture<Integer> futrue3 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@SneakyThrows@Overridepublic Integer get() {int sum = new Random().nextInt(30);Thread.sleep(2000);System.out.println(Thread.currentThread().getName() + "开始执行：" + sum);retu sum;}},executorService);CompletableFuture<Object> result = CompletableFuture.anyOf(futrue1, futrue2, futrue3);System.out.println(result.get());}

　　5.allOf

　　　　多个 CompletableFuture 的同时返回　　

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)

　

 

 @SneakyThrowspublic static void allOfDemo(){ExecutorService executorService = Executors.newFixedThreadPool(3);CompletableFuture<Integer> futrue1 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@SneakyThrows@Overridepublic Integer get() {int sum = new Random().nextInt(30);Thread.sleep(3000);System.out.println(Thread.currentThread().getName() + "开始执行：" + sum);retu sum;}},executorService);CompletableFuture<Integer> futrue2 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@SneakyThrows@Overridepublic Integer get() {int sum = new Random().nextInt(30);Thread.sleep(5000);System.out.println(Thread.currentThread().getName() + "开始执行：" + sum);retu sum;}},executorService);CompletableFuture<Integer> futrue3 = CompletableFuture.supplyAsync(new Supplier<Integer>() {@SneakyThrows@Overridepublic Integer get() {int sum = new Random().nextInt(30);Thread.sleep(2000);System.out.println(Thread.currentThread().getName() + "开始执行：" + sum);retu sum;}},executorService);CompletableFuture<Object> result = CompletableFuture.anyOf(futrue1, futrue2, futrue3);System.out.println(result.get());}

 

CompletableFuture 应用实例
　　

 

 　　这是一个烧水泡茶的程序，这里又两个线程，一个线程是T1，执行的操作是洗水壶、烧开水、等待拿茶叶之后泡茶。T2线程执行洗茶壶、洗茶杯、拿茶叶。T1在执行工序时，要等待T2执行完毕。

　　

 @SneakyThrowspublic static void TeaDemo(){ExecutorService executorService = Executors.newFixedThreadPool(3);CompletableFuture<Void> f1 = CompletableFuture.runAsync(() -> {System.out.println(Thread.currentThread().getName() + "洗水壶");try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName() + "洗水杯");try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}}, executorService);CompletableFuture<String> f2 = CompletableFuture.supplyAsync(new Supplier<String>() {@Overridepublic String get() {System.out.println(Thread.currentThread().getName() + "洗茶壶");try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName() + "洗茶杯");try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName() + "拿茶叶");try {Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}retu "龙井";}});CompletableFuture<String> f3 = f2.thenCombine(f1, (name, _a) -> {System.out.println("拿到了茶叶" + name);System.out.println("开始泡茶");retu "龙井茶泡好了";});System.out.println(f3.get());countDownLatch.countDown();}