方法四: CyclicBarrier
CyclicBarrier的中文意思是“循环栅栏”,大意是一个可循环利用的屏障。它的作用就是会让所有线程都等待完成后才会继续下一步行动。
举个例子,就像生活中我们会约朋友们到某个餐厅一起吃饭,有些朋友可能会早到,有些朋友可能会晚到,但是这个餐厅规定必须等到所有人到齐之后才会让我们进去。这里的朋友们就是各个线程,餐厅就是CyclicBarrier。
下面简单看下CyclicBarrier的源码:
构造方法:
public CyclicBarrier(int parties) /** * parties 是参与线程的个数 * 第二个构造方法有一个 Runnable 参数,这个参数的意思是最后一个到达线程要做的任务 */ public CyclicBarrier(int parties, Runnable barrierAction)
重要方法:
/** * 线程调用 await() 表示自己已经到达栅栏 * BrokenBarrierException 表示栅栏已经被破坏,破坏的原因可能是其中一个线程 await() 时被中断或者超时 */ public int await() throws InterruptedException, BrokenBarrierException public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException
下面针对两个不同的构造方法举两个经典的示例:
import java.util.concurrent.CyclicBarrier; public class CyclicBarrierTest { public static void main(String[] args) throws Exception { long time = System.currentTimeMillis(); final int threadNum = 5; //注意:5个子线程 + 1个主线程 CyclicBarrier cb = new CyclicBarrier(threadNum + 1); for (int i = 0; i < threadNum; i++) { new Thread(new MyRunable(cb, i)).start(); } cb.await(); System.out.println("所有thread执行完成!总耗时===" + (System.currentTimeMillis() - time)); } static class MyRunable implements Runnable { CyclicBarrier barrier; int num = 0; public MyRunable(CyclicBarrier cb, int i) { this.barrier = cb; this.num = i; } @Override public void run() { try { Thread.sleep(num * 1000L); System.out.println(Thread.currentThread() + ": 耗时==" + num + " 秒" + ",正在等候其它线程完成..."); barrier.await(); } catch (Exception e) { e.printStackTrace(); } } } }
import java.util.concurrent.CyclicBarrier; public class CyclicBarrierTest2 { public static void main(String[] args) { int threadNum = 5; CyclicBarrier barrier = new CyclicBarrier(threadNum, new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName() + " 完成最后任务"); } }); for (int i = 0; i < threadNum; i++) { new TaskThread(i + "", barrier).start(); } } } class TaskThread extends Thread { String threadName; CyclicBarrier barrier; public TaskThread(String name, CyclicBarrier barrier) { this.threadName = name; this.barrier = barrier; } @Override public void run() { try { Thread.sleep(1000); /** * 使用CyclicBarrier时要注意异常情况,一旦没处理好导致cyclicBarrier.await()没执行会引起线程阻塞,导致CPU居高不下**/ /*if ("3".equals(threadName)) { System.out.println(Integer.parseInt("1.233")); }*/ System.out.println("线程:" + threadName + " 到达栅栏 A"); barrier.await(); System.out.println("线程:" + threadName + " 冲破栅栏 A"); Thread.sleep(2000); System.out.println("线程:" + threadName + " 到达栅栏 B"); barrier.await(); System.out.println("线程:" + threadName + " 冲破栅栏 B"); } catch (Exception e) { e.printStackTrace(); } } }
第二个示例,运行日志如下:
线程:1 到达栅栏 A 线程:0 到达栅栏 A 线程:3 到达栅栏 A 线程:4 到达栅栏 A 线程:2 到达栅栏 A Thread-2 完成最后任务 线程:2 冲破栅栏 A 线程:1 冲破栅栏 A 线程:3 冲破栅栏 A 线程:4 冲破栅栏 A 线程:0 冲破栅栏 A 线程:4 到达栅栏 B 线程:3 到达栅栏 B 线程:1 到达栅栏 B 线程:0 到达栅栏 B 线程:2 到达栅栏 B Thread-2 完成最后任务 线程:2 冲破栅栏 B 线程:4 冲破栅栏 B 线程:3 冲破栅栏 B 线程:1 冲破栅栏 B 线程:0 冲破栅栏 B Process finished with exit code 0
从打印结果可以看出,所有线程会等待全部线程到达栅栏之后才会继续执行,并且最后到达的线程会完成 Runnable 的任务。看得出来这种方式,比较适合用于多线程计算数据,最后合并计算结果的场景。
另外对比一下,CyclicBarrier 与 CountDownLatch的区别:
- CountDownLatch 是一次性的,CyclicBarrier 是可循环利用的
- CountDownLatch 参与的线程的职责是不一样的,有的在倒计时,有的在等待倒计时结束。CyclicBarrier 参与的线程职责是一样的。
另外,可以看到使用CyclicBarrier也是有一定风险的,如果运行中没有捕获相关异常很容易导致CPU居高不下从而导致整个项目无法运行(想测试的同学可以把注解打开),那么遇到这种问题如何处理,当然把整个代码try catch是一种解决方式。
另外一种比较优雅的解决方式是使用cyclicBarrier.await(5, TimeUnit.SECONDS) 代替cyclicBarrier.await(),方法中的那2个参数分别是超时时间和超时单位,如果线程在规定的时间内没有处理完成则主线程被自动唤醒继续执行下一步操作。
方法五: Future
JDK5新增了Future接口,用于描述一个异步计算的结果。
虽然 Future 以及相关使用方法提供了异步执行任务的能力,但是对于结果的获取却是很不方便,只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和我们的异步编程的初衷相违背,轮询的方式又会耗费无谓的 CPU 资源,而且也不能及时地得到计算结果。
Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果等操作。必要时可以通过get方法获取执行结果,该方法会阻塞直到任务返回结果。
Future类位于java.util.concurrent包下,它是一个接口:
public interface Future<V> { /** * 方法用来取消任务,如果取消任务成功则返回true,如果取消任务失败则返回false。 * * @param mayInterruptIfRunning 表示是否允许取消正在执行却没有执行完毕的任务,如果设置true,则表示可以取消正在执行过程中的任务。 * @return 如果任务已经完成,则无论mayInterruptIfRunning为true还是false,此方法肯定返回false,即如果取消已经完成的任务会返回false; * 如果任务正在执行,若mayInterruptIfRunning设置为true,则返回true,若mayInterruptIfRunning设置为false,则返回false; * 如果任务还没有执行,则无论mayInterruptIfRunning为true还是false,肯定返回true。 */ boolean cancel(boolean mayInterruptIfRunning); /** * 方法表示任务是否被取消成功 * @return 如果在任务正常完成前被取消成功,则返回 true */ boolean isCancelled(); /** * 方法表示任务是否已经完成 * @return 若任务完成,则返回true */ boolean isDone(); /** * 方法用来获取执行结果,这个方法会产生阻塞,会一直等到任务执行完毕才返回 * @return 任务执行的结果值 * @throws InterruptedException * @throws ExecutionException */ V get() throws InterruptedException, ExecutionException; /** * 用来获取执行结果,如果在指定时间内,还没获取到结果,就直接返回null(并不是抛出异常,需要注意)。 * @param timeout 超时时间 * @param unit 超时单位 * @return * @throws InterruptedException * @throws ExecutionException * @throws TimeoutException */ V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }
从上面方法的注释可以看出,Futrue提供了三种功能:
1)判断任务是否完成;
2)能够中断任务;
3)能够获取任务执行结果。(最为常用的)
下面咱就利用Future.get()来实现需求,以下例子就是一个很经典的Future的用法
package cn.huolala.sdp.fulfillment.api.controller; import java.util.ArrayList; import java.util.List; import java.util.concurrent.*; /** * @author XIAGUANGHUI * @version 1.0 * @description * @date 2022/1/10 下午2:13 */ public class FutureTest { public static void futureTest() { long time = System.currentTimeMillis(); int threadNum = 5; ExecutorService executorService = Executors.newFixedThreadPool(threadNum); List<Future<String>> list = new ArrayList<Future<String>>(threadNum); for (int i = 1; i <= threadNum; i++) { final int num = i; list.add(executorService.submit(new Callable<String>() { @Override public String call() throws Exception { Thread.sleep(num * 1000); /*if (num == 3) { System.out.println(Integer.parseInt("1.233")); }*/ return Thread.currentThread() + ": 耗时==" + num + " 秒"; } }) ); } for (Future<String> future : list) { try { System.out.println(future.get(5, TimeUnit.SECONDS)); } catch (Exception e) { e.printStackTrace(); } } System.out.println("总耗时===" + (System.currentTimeMillis() - time)); executorService.shutdownNow(); } public static void main(String[] args) { futureTest(); } }
可以看到前面几种方式的效果都是一致的,但这个方法相对前面的方法而言,代码相对复杂一点,不过有返回值。
很多博客说使用不带等待时间限制的get方法时,如果子线程执行异常了会导致主线程长期阻塞,这其实是错误的,子线程执行异常时其异常会被捕获,然后修改任务的状态为异常结束并唤醒等待的主线程,get方法判断任务状态发生变更,就终止等待了,并抛出异常。要想验证这一点,只需要把上面注释的代码放开,就可以看到效果。
方法六:使用CompletableFuture
使用Future获得异步执行结果时,要么调用阻塞方法get(),要么轮询看isDone()是否为true,这两种方法都不是很好,因为主线程也会被迫等待。
从Java 8开始引入了CompletableFuture,它针对Future做了改进,可以传入回调对象,当异步任务完成或者发生异常时,自动调用回调对象的回调方法。
我们以获取股票价格为例,看看如何使用CompletableFuture:
import java.util.concurrent.CompletableFuture; public class CompletableFutureDemo { public static void main(String[] args) throws Exception { // 创建异步执行任务: CompletableFuture<Double> cf = CompletableFuture.supplyAsync(CompletableFutureDemo::fetchPrice); // 如果执行成功: cf.thenAccept((result) -> { System.out.println("price: " + result); }); // 如果执行异常: cf.exceptionally((e) -> { e.printStackTrace(); return null; }); // 主线程不要立刻结束,否则CompletableFuture默认使用的线程池会立刻关闭: Thread.sleep(200); } static Double fetchPrice() { try { Thread.sleep(100); } catch (InterruptedException e) { } if (Math.random() < 0.3) { throw new RuntimeException("fetch price failed!"); } return 5 + Math.random() * 20; } }
创建一个CompletableFuture是通过CompletableFuture.supplyAsync()实现的,它需要一个实现了Supplier接口的对象:
public interface Supplier<T> { T get(); }
这里我们用lambda语法简化了一下,直接传入CompletableFutureDemo::fetchPrice,因为CompletableFutureDemo.fetchPrice()静态方法的签名符合Supplier接口的定义(除了方法名外)。
紧接着,CompletableFuture已经被提交给默认的线程池执行了,我们需要定义的是CompletableFuture完成时和异常时需要回调的实例。完成时,CompletableFuture会调用Consumer对象:
public interface Consumer<T> { void accept(T t); }
异常时,CompletableFuture会调用Function对象:
public interface Function<T, R> { R apply(T t); }
这里我们都用lambda语法简化了代码。
可见CompletableFuture的优点是:异步任务结束时,会自动回调某个对象的方法;异步任务出错时,会自动回调某个对象的方法;主线程设置好回调后,不再关心异步任务的执行。
如果只是实现了异步回调机制,我们还看不出CompletableFuture相比Future的优势。CompletableFuture更强大的功能是,多个CompletableFuture可以串行执行,例如,定义两个CompletableFuture,第一个CompletableFuture根据证券名称查询证券代码,第二个CompletableFuture根据证券代码查询证券价格,这两个CompletableFuture实现串行操作如下:
import java.util.concurrent.CompletableFuture; public class CompletableFutureDemo { public static void main(String[] args) throws Exception { // 第一个任务: CompletableFuture<String> cfQuery = CompletableFuture.supplyAsync(() -> { return queryCode("中国石油"); }); // cfQuery成功后继续执行下一个任务: CompletableFuture<Double> cfFetch = cfQuery.thenApplyAsync((code) -> { return fetchPrice(code); }); // cfFetch成功后打印结果: cfFetch.thenAccept((result) -> { System.out.println("price: " + result); }); // 主线程不要立刻结束,否则CompletableFuture默认使用的线程池会立刻关闭: Thread.sleep(2000); } static String queryCode(String name) { try { Thread.sleep(100); } catch (InterruptedException e) { } return "601857"; } static Double fetchPrice(String code) { try { Thread.sleep(100); } catch (InterruptedException e) { } return 5 + Math.random() * 20; } }
除了串行执行外,多个CompletableFuture还可以并行执行。例如,我们考虑这样的场景:同时从新浪和网易查询证券代码,只要任意一个返回结果,就进行下一步查询价格,查询价格也同时从新浪和网易查询,只要任意一个返回结果,就完成操作:
// CompletableFuture import java.util.concurrent.CompletableFuture; public class Main { public static void main(String[] args) throws Exception { // 两个CompletableFuture执行异步查询: CompletableFuture<String> cfQueryFromSina = CompletableFuture.supplyAsync(() -> { return queryCode("中国石油", "https://finance.sina.com.cn/code/"); }); CompletableFuture<String> cfQueryFrom163 = CompletableFuture.supplyAsync(() -> { return queryCode("中国石油", "https://money.163.com/code/"); }); // 用anyOf合并为一个新的CompletableFuture: CompletableFuture<Object> cfQuery = CompletableFuture.anyOf(cfQueryFromSina, cfQueryFrom163); // 两个CompletableFuture执行异步查询: CompletableFuture<Double> cfFetchFromSina = cfQuery.thenApplyAsync((code) -> { return fetchPrice((String) code, "https://finance.sina.com.cn/price/"); }); CompletableFuture<Double> cfFetchFrom163 = cfQuery.thenApplyAsync((code) -> { return fetchPrice((String) code, "https://money.163.com/price/"); }); // 用anyOf合并为一个新的CompletableFuture: CompletableFuture<Object> cfFetch = CompletableFuture.anyOf(cfFetchFromSina, cfFetchFrom163); // 最终结果: cfFetch.thenAccept((result) -> { System.out.println("price: " + result); }); // 主线程不要立刻结束,否则CompletableFuture默认使用的线程池会立刻关闭: Thread.sleep(200); } static String queryCode(String name, String url) { System.out.println("query code from " + url + "..."); try { Thread.sleep((long) (Math.random() * 100)); } catch (InterruptedException e) { } return "601857"; } static Double fetchPrice(String code, String url) { System.out.println("query price from " + url + "..."); try { Thread.sleep((long) (Math.random() * 100)); } catch (InterruptedException e) { } return 5 + Math.random() * 20; } }
上述逻辑实现的异步查询规则实际上是:
┌─────────────┐ ┌─────────────┐ │ Query Code │ │ Query Code │ │ from sina │ │ from 163 │ └─────────────┘ └─────────────┘ │ │ └───────┬───────┘ ▼ ┌─────────────┐ │ anyOf │ └─────────────┘ │ ┌───────┴────────┐ ▼ ▼ ┌─────────────┐ ┌─────────────┐ │ Query Price │ │ Query Price │ │ from sina │ │ from 163 │ └─────────────┘ └─────────────┘ │ │ └────────┬───────┘ ▼ ┌─────────────┐ │ anyOf │ └─────────────┘ │ ▼ ┌─────────────┐ │Display Price│ └─────────────┘
除了anyOf()可以实现“任意个CompletableFuture只要一个成功”,allOf()可以实现“所有CompletableFuture都必须成功”,这些组合操作可以实现非常复杂的异步流程控制。
最后我们注意CompletableFuture的命名规则:xxx():表示该方法将继续在已有的线程中执行;xxxAsync():表示将异步在线程池中执行。
小结:CompletableFuture可以指定异步处理流程:thenAccept()处理正常结果;
exceptional()处理异常结果;thenApplyAsync()用于串行化另一个CompletableFuture;anyOf()和allOf()用于并行化多个CompletableFuture。