24.线程系列- google提供的一些好用的并发工具类

关于并发方面，juc已帮我们提供了很多好用的工具，而谷歌在此基础上做了扩展，使并发更容易，这些工具放在guava.jar包中。

本文演示几个简单的案例，见一下guava的效果。

guava maven配置

<dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>27.0-jre</version>
</dependency>

guava中常用几个类

MoreExecutors：提供了一些静态方法，是对juc中的Executors类的一个扩展

Future：也提供了很多静态方法，是对juc中Future的一个扩展

案例1：异步执行任务完毕之后回调

public class Demo1 {public static void main(String[] args) throws ExecutionException, InterruptedException {//创建一个线程池ExecutorService executorService = Executors.newFixedThreadPool(5);ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(executorService);try {ListenableFuture<Integer> listenableFuture = listeningExecutorService.submit(() -> {System.out.println(System.currentTimeMillis());//休眠2秒，默认耗时TimeUnit.SECONDS.sleep(2);System.out.println(System.currentTimeMillis());return 10;});listenableFuture.addListener(()->{System.out.println("回调");},MoreExecutors.directExecutor());System.out.println(listenableFuture.get());}finally {listeningExecutorService.shutdown();}}
}

输出：

1608615858082
1608615860082
回调
10

另外一种写法：

public class Demo2 {public static void main(String[] args) throws ExecutionException, InterruptedException {ExecutorService delegate = Executors.newFixedThreadPool(5);try {ListeningExecutorService executorService = MoreExecutors.listeningDecorator(delegate);ListenableFuture<Integer> submit = executorService.submit(() -> {System.out.println(System.currentTimeMillis());TimeUnit.SECONDS.sleep(4);int i = 10 / 0;System.out.println(System.currentTimeMillis());return 10;});Futures.addCallback(submit, new FutureCallback<Integer>() {@Overridepublic void onSuccess(@Nullable Integer result) {System.out.println("执行成功" + result);}@Overridepublic void onFailure(Throwable t) {try {TimeUnit.MILLISECONDS.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("执行任务发生异常:" + t.getMessage());}}, MoreExecutors.directExecutor());System.out.println(submit.get());} finally {delegate.shutdown();}}
}

输出：

1608615899714
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.ArithmeticException: / by zeroat com.google.common.util.concurrent.AbstractFuture.getDoneValue(AbstractFuture.java:552)at com.google.common.util.concurrent.AbstractFuture.get(AbstractFuture.java:533)at com.google.common.util.concurrent.FluentFuture$TrustedFuture.get(FluentFuture.java:82)at com.example.thread.guava.Demo2.main(Demo2.java:40)
Caused by: java.lang.ArithmeticException: / by zeroat com.example.thread.guava.Demo2.lambda$main$0(Demo2.java:20)at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:125)at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:57)at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:78)at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)at java.lang.Thread.run(Thread.java:745)
执行任务发生异常:/ by zero

上面通过调用Futures的静态方法addCallback在异步执行的任务中添加回调，回调的对象是一个FutureCallback，此对象有2个方法，任务执行成功调用onSuccess，执行失败调用onFailure。

示例2：获取一批异步任务的执行结果

@Slf4j
public class Demo3 {public static void main(String[] args) throws ExecutionException, InterruptedException {log.info("star");ExecutorService delegate = Executors.newFixedThreadPool(5);try {ListeningExecutorService executorService = MoreExecutors.listeningDecorator(delegate);List<ListenableFuture<Integer>> futureList = new ArrayList<>();for (int i = 5; i >= 0; i--) {int j = i;futureList.add(executorService.submit(() -> {TimeUnit.SECONDS.sleep(j);return j;}));}//获取一批任务的执行结果List<Integer> resultList = Futures.allAsList(futureList).get();//输出resultList.forEach(item -> {log.info("{}", item);});} finally {delegate.shutdown();}}
}

14:26:35.970 [main] INFO com.itsoku.chat34.Demo3 - star
14:26:41.137 [main] INFO com.itsoku.chat34.Demo3 - 5
14:26:41.138 [main] INFO com.itsoku.chat34.Demo3 - 4
14:26:41.138 [main] INFO com.itsoku.chat34.Demo3 - 3
14:26:41.138 [main] INFO com.itsoku.chat34.Demo3 - 2
14:26:41.138 [main] INFO com.itsoku.chat34.Demo3 - 1
14:26:41.138 [main] INFO com.itsoku.chat34.Demo3 - 0

结果中按顺序输出了6个异步任务的结果，此处用到了Futures.allAsList方法，看一下此方法的声明：

public static <V> ListenableFuture<List<V>> allAsList(Iterable<? extends ListenableFuture<? extends V>> futures)

传递一批ListenableFuture，返回一个ListenableFuture<List<V>>，内部将一批结果转换为了一个ListenableFuture对象。

示例3：一批任务异步执行完毕之后回调


@Slf4j
public class Demo4 {public static void main(String[] args) throws ExecutionException, InterruptedException {log.info("star");ExecutorService delegate = Executors.newFixedThreadPool(5);try {ListeningExecutorService executorService = MoreExecutors.listeningDecorator(delegate);List<ListenableFuture<Integer>> futureList = new ArrayList<>();for (int i = 5; i >= 0; i--) {int j = i;futureList.add(executorService.submit(() -> {TimeUnit.SECONDS.sleep(j);return j;}));}ListenableFuture<List<Integer>> listListenableFuture = Futures.allAsList(futureList);Futures.addCallback(listListenableFuture, new FutureCallback<List<Integer>>() {@Overridepublic void onSuccess(@Nullable List<Integer> result) {log.info("result中所有结果之和：" + result.stream().reduce(Integer::sum).get());}@Overridepublic void onFailure(Throwable t) {log.error("执行任务发生异常:" + t.getMessage(), t);}}, MoreExecutors.directExecutor());} finally {delegate.shutdown();}}
}

输出：

14:47:04.819 [main] INFO com.itsoku.chat34.Demo4 - star
14:47:09.933 [pool-1-thread-1] INFO com.itsoku.chat34.Demo4 - result中所有结果之和：15

代码中异步执行了一批任务，所有任务完成之后，回调了上面的onSuccess方法，内部对所有的结果进行sum操作。

总结

通过guava提供的一些工具类，方便异步执行任务并进行回调
guava内部还有很多好用的工具类，有兴趣的可以去研究一下