比如主线程让一个子线程去执行任务#xff0c;子线…Future

接口理论知识复习

实现类定义了操作异步任务执行的一些方法如获取异步任务的执行结果、取消任务的执行、判断任务是否被取消、判断任务执行是否完毕等。

比如主线程让一个子线程去执行任务子线程可能比较耗时启动子线程开始执行任务后主线程就去做其他事情了忙其他事情或者先执行完过了一会才去获取子任务的执行结果或变更的任务状态。

总结Future

接口可以为主线程开一个分支任务专门为主线程处理耗时和复杂业务。

Future

把这个任务放到异步线程中执行。

主线程继续处理其他任务或者先行结束再通过

Future

Thread(futureTask);t1.start();System.out.println(futureTask.get());}

}class

{method1();System.out.println(------------------);method2();}private

static

Executors.newFixedThreadPool(3);long

startTime

System.currentTimeMillis();FutureTaskString

futureTask1

{TimeUnit.MILLISECONDS.sleep(500);}

catch

over;});executorService.submit(futureTask1);FutureTaskString

futureTask2

{TimeUnit.MILLISECONDS.sleep(300);}

catch

over;});executorService.submit(futureTask2);FutureTaskString

futureTask3

{TimeUnit.MILLISECONDS.sleep(300);}

catch

over;});executorService.submit(futureTask3);long

endTime

System.currentTimeMillis();System.out.println(----costTime:

(endTime

毫秒);System.out.println(Thread.currentThread().getName()

----end);executorService.shutdown();}private

static

{TimeUnit.MILLISECONDS.sleep(500);}

catch

{TimeUnit.MILLISECONDS.sleep(300);}

catch

{TimeUnit.MILLISECONDS.sleep(300);}

catch

System.currentTimeMillis();System.out.println(----costTime:

(endTime

毫秒);System.out.println(Thread.currentThread().getName()

----end);}

{log.debug(Thread.currentThread().getName()

-----come

t1);t1.start();log.debug(futureTask.get());log.debug(Thread.currentThread().getName()

20:59:50.067

方法求结果如果计算没有完成容易导致程序阻塞。

如果不希望等待过长时间希望在等待指定时间后自动结束。

可以使用

get(long

{log.debug(Thread.currentThread().getName()

-----come

t1);t1.start();log.debug(Thread.currentThread().getName()

----忙其他任务了);log.debug(futureTask.get(2,

TimeUnit.SECONDS));}

{log.debug(Thread.currentThread().getName()

-----come

t1);t1.start();log.debug(Thread.currentThread().getName()

{log.debug(futureTask.get());break;}

else

{TimeUnit.MILLISECONDS.sleep(500);}

catch

RuntimeException(e);}log.debug(正在处理中...);}}}

22:56:26.267

资源而且也不见得能及时地得到计算结果。

如果想要异步获取结果通常都会以轮询的方式去获取结果尽量不要阻塞。

Future

对于结果的获取不是很友好只能通过阻塞或轮询的方式得到任务的结果。

Future

的。

但是对于回调通知通过轮询的方式去判断任务是否完成这样非常占用

一些复杂任务

将多个异步任务的计算结果组合起来后一个异步任务的计算结果需要前一个异步任务的值。

将这两个或多个异步计算合成一个一步计算这几个一步计算相互独立同时后面这个又依赖前一个处理的结果。

对计算速度选最快当Future集合中某个任务最快结束时返回结果返回第一名处理

对于简单的业务场景使用Future完全OK,但想完成上述一些复杂的任务使用Future之前提供的那点API就囊中羞涩处理起来不够优雅这时候还是让CompletableFuture以声明式的方式优雅的处理这些需求。

Future能干的CompletableFuture都能干。

CompletableFuture

CompletableFuture。

CompletableFuture

提供了一种观察者模式类似的机制可以让人物执行完成后通知监听的一方。

CompletableFuture

代表异步计算过程中的某一个阶段一个阶段完成以后可能会触发另外一个阶段一个阶段的计算执行可以是一个

FunctionConsumer

Systemm.out.print(x)).thenRun(()

System.out.println())一个阶段的执行可能是被单个阶段的完成触发也可能是由多个阶段一起触发。

CompletableFuture

的扩展功能可以帮助我们简化异步编程的复杂性并且提供了函数式编程的能力可以通过回调的方式处理计算结果也提供了转换和组合

CompletableFuture

Future也有可能代表一个完成阶段CompletionStage它支持在计算完成以后触发一些函数或执行某些动作。

它实现了

Future

asyncRunStage(screenExecutor(executor),

runnable);

asyncSupplyStage(screenExecutor(executor),

supplier);

没有指定Executor的方法直接使用默认的ForkJoinPool.commonPool()作为它的线程池执行异步代码。

如果指定线程池则使用我们自定义的或者特别指定的线程池执行异步代码。

代码演示

{log.debug(Thread.currentThread().getName()

------come

RuntimeException(e);}log.debug(--------task

over);});System.out.println(future.get());}

}输出

Executors.newFixedThreadPool(3);CompletableFutureVoid

future

{log.debug(Thread.currentThread().getName()

------come

RuntimeException(e);}log.debug(--------task

over);},

executorService);System.out.println(future.get());executorService.shutdown();}

}输出

CompletableFuture.supplyAsync(()

{log.debug(Thread.currentThread().getName()

------come

RuntimeException(e);}log.debug(--------task

over);return

supplyAsync;});log.debug(future.get());}

}输出

Executors.newFixedThreadPool(3);CompletableFutureString

future

CompletableFuture.supplyAsync(()

{log.debug(Thread.currentThread().getName()

------come

RuntimeException(e);}log.debug(--------task

over);return

executorService);log.debug(future.get());executorService.shutdown();}

}输出

的功能增强版减少阻塞和轮询可以传入回调对象当异步任务完成或者发生异常时自动调用回调对象的回调方法。

Slf4j(topic

{CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread().getName()

----come

ThreadLocalRandom.current().nextInt(10);try

catch

RuntimeException(e);}System.out.println(----1秒钟后出结果

result);return

{System.out.println(----计算完成更新系统UpdateValue

{e.printStackTrace();System.out.println(异常情况

e.getCause()

null;});System.out.println(Thread.currentThread().getName()

线程先去忙其他任务);}

线程池中的线程类似于守护线程由于主线程执行速度过快先执行结束导致

Fork/Join

Executors.newFixedThreadPool(3);CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread().getName()

----come

ThreadLocalRandom.current().nextInt(10);try

catch

RuntimeException(e);}System.out.println(----1秒钟后出结果

result);return

{System.out.println(----计算完成更新系统UpdateValue:

{e.printStackTrace();System.out.println(异常情况

e.getCause()

null;});System.out.println(Thread.currentThread().getName()

}输出

接口可以接收两个参数其中第一个参数为上一步中产生的结果第二个参数为上一步代码执行过程中产生的异常。

exceptionally

exceptionally(FunctionThrowable,

extends

Executors.newFixedThreadPool(3);CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread().getName()

----come

ThreadLocalRandom.current().nextInt(10);try

catch

RuntimeException(e);}System.out.println(----1秒钟后出结果

result);if(result

{System.out.println(----计算完成更新系统UpdateValue:

{e.printStackTrace();System.out.println(异常情况

e.getCause()

null;});System.out.println(Thread.currentThread().getName()

}输出

异步任务结束时会自动回调某个对象的方法主线程设置好回调后不再关心异步任务的执行异步任务之间可以顺序执行异步任务出错时会自动回调某个对象的方法

案例精讲

函数式接口名称方法名称参数返回值Runnablerun无参数无返回值Functionapply1

个参数无返回值Supplierget无参数有返回值BiConsumeraccept2

个参数无返回值

同一款产品同时搜索出同款产品在各大电商平台的售价同一款产品同时搜索出本产品在同一个电商平台下各个入驻卖家售价是多少

输出返回

出来结果希望是同款产品的在不同地方的价格清单列表返回一个List

例如《Mysql》

%.2f,netMall.getNetMallName(),netMall.calPrice(productName))).collect(Collectors.toList());}public

static

getPriceByCompletableFuture(ListNetMall

list,

-CompletableFuture.supplyAsync(()

price

%.2f,netMall.getNetMallName(),netMall.calPrice(productName)))).collect(Collectors.toList()).stream().map(s

s.join()).collect(Collectors.toList());}public

static

getPriceByCompletableFuture(list,

mysql);for

{System.out.println(element);}long

endTime

System.currentTimeMillis();System.out.println(----costTime:

(endTime

ThreadLocalRandom.current().nextDouble()

getPrice()

计算完成就返回正常值否则返回备胎值传入的参数立即获取结果不阻塞

public

CompletableFuture.supplyAsync(()

{try

RuntimeException(e);}System.out.println(future.getNow(xxx));System.out.println(future.complete(completeValue)

future.join());}

计算结果存在依赖关系两个线程串行化由于存在依赖关系当前步骤出错不再执行下一步当前步骤有异常的话就停止运行。

Slf4j(topic

Executors.newFixedThreadPool(3);CompletableFuture.supplyAsync(()

{try

RuntimeException(e);}System.out.println(111);return

1;},

{System.out.println(222);return

2;}).thenApply(f

{System.out.println(333);return

{if(e

{e.printStackTrace();;System.out.println(e.getMessage());return

null;});System.out.println(Thread.currentThread().getName()

线程先去忙其他任务);}

Executors.newFixedThreadPool(3);CompletableFuture.supplyAsync(()

{try

RuntimeException(e);}System.out.println(111);return

1;},

0;System.out.println(222);return

2;}).thenApply(f

{System.out.println(333);return

{if(e

{e.printStackTrace();;System.out.println(e.getMessage());return

null;});System.out.println(Thread.currentThread().getName()

}输出结果

计算结果存在依赖关系两个线程串行化有异常也可以继续向下运行根据带的异常参数可以进一步处理

如下代码中步骤

Executors.newFixedThreadPool(3);CompletableFuture.supplyAsync(()

{try

RuntimeException(e);}System.out.println(111);return

1;},

0;System.out.println(222);return

2;}).handle((f,

{System.out.println(333);return

{if(e

{e.printStackTrace();;System.out.println(e.getMessage());return

null;});System.out.println(Thread.currentThread().getName()

}输出结果

{CompletableFuture.supplyAsync(()

{return

{System.out.println(CompletableFuture.supplyAsync(()

resultA)//

{}).join());System.out.println(--------------------------------);System.out.println(CompletableFuture.supplyAsync(()

resultA)//

{System.out.println(r);}).join());System.out.println(--------------------------------);System.out.println(CompletableFuture.supplyAsync(()

resultA)//

ForkJoinPool传入了一个自定义线程池如果执行第一个任务的时候传入了一个自定义线程池

theRun

方法执行第二个任务时则第二个任务和第一个任务是共用同一个线程池调用

theRunAsync

执行第二个任务时则第一个人任务使用的是你自己传入的线程池第二个任务使用的是

ForkJoinPool

CompletableFuture.supplyAsync(()

{try

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(1号任务\t

Thread.currentThread().getName());return

abcd;}).thenRun(()

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(2号任务\t

Thread.currentThread().getName());}).thenRun(()

{try

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(3号任务\t

Thread.currentThread().getName());}).thenRun(()

{try

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(4号任务\t

Thread.currentThread().getName());});System.out.println(future.get(2L,

TimeUnit.SECONDS));}

Executors.newFixedThreadPool(3);CompletableFutureVoid

future

CompletableFuture.supplyAsync(()

{try

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(1号任务\t

Thread.currentThread().getName());return

abcd;},

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(2号任务\t

Thread.currentThread().getName());}).thenRun(()

{try

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(3号任务\t

Thread.currentThread().getName());}).thenRun(()

{try

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(4号任务\t

Thread.currentThread().getName());});System.out.println(future.get(2L,

TimeUnit.SECONDS));pool.shutdown();}

}输出结果

Executors.newFixedThreadPool(3);CompletableFutureVoid

future

CompletableFuture.supplyAsync(()

{try

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(1号任务\t

Thread.currentThread().getName());return

abcd;},

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(2号任务\t

Thread.currentThread().getName());}).thenRunAsync(()

{try

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(3号任务\t

Thread.currentThread().getName());}).thenRunAsync(()

{try

{TimeUnit.MILLISECONDS.sleep(20);}

catch

RuntimeException(e);}System.out.println(4号任务\t

Thread.currentThread().getName());});System.out.println(future.get(2L,

TimeUnit.SECONDS));pool.shutdown();}

}输出结果

CompletableFuture.supplyAsync(()

come

planA;});CompletableFutureString

planB

CompletableFuture.supplyAsync(()

come

planB;});CompletableFutureString

result

winer;});System.out.println(Thread.currentThread().getName()

\t----

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread().getName()

\t----启动);try

CompletableFuture.supplyAsync(()

{System.out.println(Thread.currentThread().getName()

\t----启动);try

{System.out.println(-----开始两个结果的合并);return

y;});System.out.println(combine.join());}

}输出结果