国内开源代码网站,焦作做网站哪家好,化妆品网站建设规划书范文,浙江众安建设集团有限公司网站JUC中工具类CompletableFutureCompletableFuture是java8中新增的一个类#xff0c;算是对Future的一种增强#xff0c;用起来很方便#xff0c;也是会经常用到的一个工具类#xff0c;熟悉一下。CompletionStage接口CompletionStage代表异步计算过程中的某一个阶段#xf…JUC中工具类CompletableFutureCompletableFuture是java8中新增的一个类算是对Future的一种增强用起来很方便也是会经常用到的一个工具类熟悉一下。CompletionStage接口CompletionStage代表异步计算过程中的某一个阶段一个阶段完成以后可能会触发另外一个阶段一个阶段的计算执行可以是一个FunctionConsumer或者Runnable。比如stage.thenApply(x - square(x)).thenAccept(x - System.out.print(x)).thenRun(() - System.out.println())一个阶段的执行可能是被单个阶段的完成触发也可能是由多个阶段一起触发CompletableFuture类在Java8中CompletableFuture提供了非常强大的Future的扩展功能可以帮助我们简化异步编程的复杂性并且提供了函数式编程的能力可以通过回调的方式处理计算结果也提供了转换和组合 CompletableFuture 的方法。它可能代表一个明确完成的Future也有可能代表一个完成阶段( CompletionStage )它支持在计算完成以后触发一些函数或执行某些动作。它实现了Future和CompletionStage接口常见的方法熟悉一下runAsync 和 supplyAsync方法CompletableFuture 提供了四个静态方法来创建一个异步操作。public static CompletableFuture runAsync(Runnable runnable)public static CompletableFuture runAsync(Runnable runnable, Executor executor)public static CompletableFuture supplyAsync(Supplier supplier)public static CompletableFuture supplyAsync(Supplier supplier, Executor executor)没有指定Executor的方法会使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。如果指定线程池则使用指定的线程池运行。以下所有的方法都类同。runAsync方法不支持返回值。supplyAsync可以支持返回值。示例public class CompletableFutureTest1 {public static void main(String[] args) throws Exception {CompletableFutureTest1.runAsync();CompletableFutureTest1.supplyAsync();}//runAsync方法不支持返回值 Runnablepublic static void runAsync() throws Exception {CompletableFuture future CompletableFuture.runAsync(() - {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {}System.out.println(run end ...);});future.get();}//supplyAsync可以支持返回值 Supplierpublic static void supplyAsync() throws Exception {CompletableFuture future CompletableFuture.supplyAsync(() - {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {}System.out.println(run end ...);return System.currentTimeMillis();});//如果没有future.get()阻塞等待结果的话因为CompletableFuture.supplyAsync()方法默认是守护线程形式执行任务在主线程结束后会跟着退出// 如果传入的是线程池去执行这不是守护线程不会导致退出long time future.get();System.out.println(time time);}}输出run end ...run end ...time 1599556248764计算结果完成时的回调方法当CompletableFuture的计算结果完成或者抛出异常的时候可以执行特定的Action。主要是下面的方法public CompletableFuture whenComplete(BiConsumer super T,? super Throwable action)public CompletableFuture whenCompleteAsync(BiConsumer super T,? super Throwable action)public CompletableFuture whenCompleteAsync(BiConsumer super T,? super Throwable action, Executor executor)public CompletableFuture exceptionally(Function fn)可以看到Action的类型是BiConsumer它可以处理正常的计算结果或者异常情况。whenComplete 和 whenCompleteAsync 的区别whenComplete当前任务的线程继续执行 whenComplete 的任务。whenCompleteAsync把 whenCompleteAsync 这个任务继续提交给线程池来进行执行。示例public class CompletableFutureTest1 {public static void main(String[] args) throws Exception {CompletableFutureTest1.whenComplete();CompletableFutureTest1.whenCompleteBySupply();}public static void whenComplete() throws Exception {CompletableFuture future CompletableFuture.runAsync(() - {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {}if (new Random().nextInt() % 2 0) {int i 12 / 0;//run end ...//执行完成//int i 12 / 0;}System.out.println(run end ...);});//对执行成功或者执行异常做处理的回调方法future.whenComplete((avoid, throwable) - {//先判断是否抛异常分开处理if (throwable ! null) {System.out.println(执行失败 throwable.getMessage());} else {System.out.println(执行完成);}});//对执行异常做处理的回调方法future.exceptionally(throwable - {System.out.println(执行失败 throwable.getMessage());return null;});TimeUnit.SECONDS.sleep(2);}public static void whenCompleteBySupply() throws Exception {CompletableFuture future CompletableFuture.supplyAsync(() - {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {}if (new Random().nextInt() % 2 0) {//int i 12 / 0;//run end ...//执行完成//int i 12 / 0;}System.out.println(run end ...);return success;});//whenComplete在thenAccept之前执行future.thenAccept(result - {System.out.println(result);});//对执行成功或者执行异常做处理的回调方法future.whenComplete((avoid, throwable) - {//先判断是否抛异常分开处理if (throwable ! null) {System.out.println(执行失败 throwable.getMessage());} else {System.out.println(执行完成);}});//对执行异常做处理的回调方法future.exceptionally(throwable - {System.out.println(执行失败 throwable.getMessage());return null;});TimeUnit.SECONDS.sleep(2);}}输出执行失败java.lang.ArithmeticException: / by zero执行失败java.lang.ArithmeticException: / by zerorun end ...执行完成successthenApply 方法当一个线程依赖另一个线程时可以使用 thenApply、thenApplyAsync 方法来把这两个线程串行化。public CompletableFuture thenApply(Function super T,? extends U fn)public CompletableFuture thenApplyAsync(Function super T,? extends U fn)public CompletableFuture thenApplyAsync(Function super T,? extends U fn, Executor executor)Function super T,? extends UT上一个任务返回结果的类型U当前任务的返回值类型示例public class CompletableFutureTest2 {public static void main(String[] args) throws Exception {CompletableFutureTest2.thenApply();}//多个CompletableFuture可以串行执行//当一个线程依赖另一个线程时可以使用 thenApply 方法来把这两个线程串行化。//多个任务串行执行,第二个任务依赖第一个任务的结果。private static void thenApply() throws Exception {CompletableFuture future CompletableFuture.supplyAsync(() - {long result new Random().nextInt(100);System.out.println(result1 result);return result;}).thenApply((t - {long result t * 5;System.out.println(result2 result);return result;}));//方式一阻塞等待结果long result future.get();System.out.println(result2: result);//方式二调用成功后接收任务的处理结果并消费处理无返回结果future.thenAccept((r) - {System.out.println(result2: r);});}}输出result141result2205result2: 205result2: 205handle 方法——可以处理正常和异常情况的thenApply 方法handle 是执行任务完成时对结果的处理。handle 方法和 thenApply 方法处理方式基本一样。不同的是 handle 是在任务完成后再执行还可以处理异常的任务。thenApply 只可以执行正常的任务任务出现异常则不执行 thenApply 方法。public CompletionStage handle(BiFunction super T, Throwable, ? extends U fn);public CompletionStage handleAsync(BiFunction super T, Throwable, ? extends U fn);public CompletionStage handleAsync(BiFunction super T, Throwable, ? extends U fn,Executor executor);示例在 handle 中可以根据任务是否有异常来进行做相应的后续处理操作。而 thenApply 方法如果上个任务出现错误则不会执行 thenApply 方法。public class CompletableFutureTest3 {public static void main(String[] args) throws Exception {CompletableFutureTest3.handle();}public static void handle() throws Exception {CompletableFuture future CompletableFuture.supplyAsync(new Supplier() {Overridepublic Integer get() {int i 10 / 0;return new Random().nextInt(10);}}).handle((param, throwable) - {int result -1;if (throwable null) {result param * 2;} else {System.out.println(throwable.getMessage());}return result;}/*new BiFunction() {Overridepublic Integer apply(Integer param, Throwable throwable) {int result -1;if(throwablenull){result param * 2;}else{System.out.println(throwable.getMessage());}return result;}}*/);System.out.println(future.get());}}输出java.lang.ArithmeticException: / by zero-1thenAccept 消费处理结果——无返回结果接收任务的处理结果并消费处理无返回结果。public CompletionStage thenAccept(Consumer super T action);public CompletionStage thenAcceptAsync(Consumer super T action);public CompletionStage thenAcceptAsync(Consumer super T action,Executor executor);示例public class CompletableFutureTest3 {public static void main(String[] args) throws Exception {//CompletableFutureTest3.handle();CompletableFutureTest3.thenAccept();}public static void thenAccept() throws Exception {CompletableFuture future CompletableFuture.supplyAsync(() - {return new Random().nextInt(10);}).thenAccept(integer - {System.out.println(integer);});future.get();}}//输出5thenRun 方法——继续执行下一个Runnable任务不获取上一个任务的处理结果跟 thenAccept 方法不一样的是不关心任务的处理结果。只要上面的任务执行完成就开始执行 thenRun 。public CompletionStage thenRun(Runnable action);public CompletionStage thenRunAsync(Runnable action);public CompletionStage thenRunAsync(Runnable action,Executor executor);示例public class CompletableFutureTest3 {public static void main(String[] args) throws Exception {CompletableFutureTest3.thenRun();}public static void thenRun() throws Exception{CompletableFuture future CompletableFuture.supplyAsync(new Supplier() {Overridepublic Integer get() {try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}return new Random().nextInt(10);}}).thenRun(() - {System.out.println(thenRun ...);});future.get();}}//2秒后输出thenRun ...thenCombine 合并任务thenCombine 会把 两个 CompletionStage 的任务都执行完成后把两个任务的结果一块交给 thenCombine 来处理。public CompletionStage thenCombine(CompletionStage extends U other,BiFunction super T,? super U,? extends V fn);public CompletionStage thenCombineAsync(CompletionStage extends U other,BiFunction super T,? super U,? extends V fn);public CompletionStage thenCombineAsync(CompletionStage extends U other,BiFunction super T,? super U,? extends V fn,Executor executor);示例public class CompletableFutureTest3 {public static void main(String[] args) throws Exception {CompletableFutureTest3.thenCombine();}private static void thenCombine() throws Exception {CompletableFuture future1 CompletableFuture.supplyAsync(() - {return hello;});CompletableFuture future2 CompletableFuture.supplyAsync(() - {return world;});CompletableFuture result future1.thenCombine(future2, (result1, result2) - {return result1 result2;});System.out.println(result.get());}}//输出hello worldthenAcceptBoth当两个CompletionStage都执行完成后把结果一块交给thenAcceptBoth来进行消耗。public CompletionStage thenAcceptBoth(CompletionStage extends U other,BiConsumer super T, ? super U action);public CompletionStage thenAcceptBothAsync(CompletionStage extends U other,BiConsumer super T, ? super U action);public CompletionStage thenAcceptBothAsync(CompletionStage extends U other,BiConsumer super T, ? super U action, Executor executor);示例public class CompletableFutureTest3 {public static void main(String[] args) throws Exception {CompletableFutureTest3.thenAcceptBoth();//等待守护进程执行完TimeUnit.SECONDS.sleep(5);}private static void thenAcceptBoth() throws Exception {CompletableFuture f1 CompletableFuture.supplyAsync(() - {int t new Random().nextInt(3);try {TimeUnit.SECONDS.sleep(t);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(f1 t);return t;});CompletableFuture f2 CompletableFuture.supplyAsync(() - {int t new Random().nextInt(3);try {TimeUnit.SECONDS.sleep(t);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(f2 t);return t;});f1.thenAcceptBoth(f2, (result1, result2) - {System.out.println(f1 result1 ;f2 result2 ;);});}}输出f11f21f11;f21;applyToEither 方法——有返回值消耗两个CompletionStage谁执行返回的结果快我就用那个CompletionStage的结果进行下一步的转化操作。public CompletionStage applyToEither(CompletionStage extends T other,Function super T, U fn);public CompletionStage applyToEitherAsync(CompletionStage extends T other,Function super T, U fn);public CompletionStage applyToEitherAsync(CompletionStage extends T other,Function super T, U fn,Executor executor);示例public class CompletableFutureTest3 {public static void main(String[] args) throws Exception {CompletableFutureTest3.applyToEither();}private static void applyToEither() throws Exception {CompletableFuture f1 CompletableFuture.supplyAsync(()-{int t 1;try {TimeUnit.SECONDS.sleep(t);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(f1t);return t;});CompletableFuture f2 CompletableFuture.supplyAsync(()-{int t 2;try {TimeUnit.SECONDS.sleep(t);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(f2t);return t;});CompletableFuture result f1.applyToEither(f2, (r)-{System.out.println(r);return r * 2;});System.out.println(result.get());}输出f1112acceptEither 方法——无返回值消耗两个CompletionStage谁执行返回的结果快我就用那个CompletionStage的结果进行下一步的消耗操作。注意这时候其实两个CompletionStage都是会执行完的只是我们只获取其中的一个比较快的结果而已参考示例的输出。public CompletionStage acceptEither(CompletionStage extends T other,Consumer super T action);public CompletionStage acceptEitherAsync(CompletionStage extends T other,Consumer super T action);public CompletionStage acceptEitherAsync(CompletionStage extends T other,Consumer super T action,Executor executor);示例public class CompletableFutureTest3 {public static void main(String[] args) throws Exception {//CompletableFutureTest3.applyToEither();CompletableFutureTest3.acceptEither();TimeUnit.SECONDS.sleep(5);}private static void acceptEither() throws Exception {CompletableFuture f1 CompletableFuture.supplyAsync(() - {int t new Random().nextInt(3);try {TimeUnit.SECONDS.sleep(t);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(f1 t);return t;});CompletableFuture f2 CompletableFuture.supplyAsync(() - {int t new Random().nextInt(3);try {TimeUnit.SECONDS.sleep(t);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(f2 t);return t;});f1.acceptEither(f2, (t) - {System.out.println(t);});}}输出f111f22runAfterEither 方法两个CompletionStage任何一个完成了都会执行下一步的操作(Runnable),两个CompletionStage都是会执行完的.public CompletionStage runAfterEither(CompletionStage other,Runnable action);public CompletionStage runAfterEitherAsync(CompletionStage other,Runnable action);public CompletionStage runAfterEitherAsync(CompletionStage other,Runnable action,Executor executor);示例代码public class CompletableFutureTest3 {public static void main(String[] args) throws Exception {//CompletableFutureTest3.applyToEither();//CompletableFutureTest3.acceptEither();CompletableFutureTest3.runAfterEither();TimeUnit.SECONDS.sleep(5);}private static void runAfterEither() throws Exception {CompletableFuture f1 CompletableFuture.supplyAsync(new Supplier() {Overridepublic Integer get() {int t new Random().nextInt(3);try {TimeUnit.SECONDS.sleep(t);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(f1 t);return t;}});CompletableFuture f2 CompletableFuture.supplyAsync(new Supplier() {Overridepublic Integer get() {int t new Random().nextInt(3);try {TimeUnit.SECONDS.sleep(t);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(f2 t);return t;}});f1.runAfterEither(f2, ()-{System.out.println(上面有一个已经完成了。);});}}输出f10上面有一个已经完成了。f21runAfterBoth两个CompletionStage都完成了计算才会执行下一步的操作(Runnable)注意输出顺序runAfterBoth方法要等两个CompletionStage都执行完了才会执行。public CompletionStage runAfterBoth(CompletionStage other,Runnable action);public CompletionStage runAfterBothAsync(CompletionStage other,Runnable action);public CompletionStage runAfterBothAsync(CompletionStage other,Runnable action,Executor executor);示例代码public class CompletableFutureTest3 {public static void main(String[] args) throws Exception {//CompletableFutureTest3.applyToEither();//CompletableFutureTest3.acceptEither();//CompletableFutureTest3.runAfterEither();CompletableFutureTest3.runAfterBoth();TimeUnit.SECONDS.sleep(5);}private static void runAfterBoth() throws Exception {CompletableFuture f1 CompletableFuture.supplyAsync(new Supplier() {Overridepublic Integer get() {int t new Random().nextInt(3);try {TimeUnit.SECONDS.sleep(t);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(f1t);return t;}});CompletableFuture f2 CompletableFuture.supplyAsync(new Supplier() {Overridepublic Integer get() {int t new Random().nextInt(3);try {TimeUnit.SECONDS.sleep(t);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(f2t);return t;}});f1.runAfterBoth(f2, new Runnable() {Overridepublic void run() {System.out.println(上面两个任务都执行完成了。);}});}}输出f11f22上面两个任务都执行完成了。thenCompose 方法thenCompose 方法允许你对两个 CompletionStage 进行流水线操作第一个操作完成时将其结果作为参数传递给第二个操作。public CompletableFuture thenCompose(Function super T, ? extends CompletionStage fn);public CompletableFuture thenComposeAsync(Function super T, ? extends CompletionStage fn) ;public CompletableFuture thenComposeAsync(Function super T, ? extends CompletionStage fn, Executor executor) ;示例代码public class CompletableFutureTest3 {public static void main(String[] args) throws Exception {CompletableFutureTest3.thenCompose();TimeUnit.SECONDS.sleep(3);}private static void thenCompose() throws Exception {CompletableFuture f CompletableFuture.supplyAsync(() - {int t new Random().nextInt(3);System.out.println(t1 t);return t;}).thenCompose((param) - {return CompletableFuture.supplyAsync(() - {int t param * 2;System.out.println(t2 t);return t;});});System.out.println(thenCompose result : f.get());}}输出t11t22thenCompose result : 2疑问Q:thenAcceptBoth与thenCombine 的区别是什么A:thenAcceptBoth无返回值消耗执行thenCombine 会有返回值。一般accept都是没有返回值的apply是有返回值的。Q:thenCompose 与thenApply 方法 的区别是什么不都是串行执行下一个任务并把第一个任务作为参数传递给第二个任务么获取线程执行结果的6种方法方式1Thread的join()方法实现代码中通过join方式阻塞了当前主线程当thread线程执行完毕之后join方法才会继续执行。join的方式只能阻塞一个线程如果其他线程中也需要获取thread线程的执行结果join方法无能为力了。示例public class ThreadJoinTest {//用于封装结果static class Result {T result;public T getResult() {return result;}public void setResult(T result) {this.result result;}}public static void main(String[] args) throws InterruptedException {Result result new Result();Thread t new Thread(() - {System.out.println(start thread!);try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}result.setResult(success);System.out.println(end thread!);});t.start();//让主线程等待thread线程执行完毕之后再继续join方法会让当前线程阻塞t.join();System.out.println(main get resultresult.getResult());}}输出start thread!end thread!main get resultsuccess方式2CountDownLatch实现使用CountDownLatch可以让一个或者多个线程等待一批线程完成之后自己再继续.示例public class CountDownLatchTest2 {static class Result{private T result;public T getResult() {return result;}public void setResult(T result) {this.result result;}}public static void main(String[] args) throws InterruptedException {Result result new Result();CountDownLatch latch new CountDownLatch(1);Thread t new Thread(() - {System.out.println(start thread!);try {TimeUnit.SECONDS.sleep(1);result.setResult(success);System.out.println(end thread!);} catch (InterruptedException e) {e.printStackTrace();}finally {latch.countDown();}});t.start();latch.await();System.out.println(main get resultresult.getResult());}}输出start thread!end thread!main get resultsuccess方式3ExecutorService.submit方法实现——ThreadPoolExecutor使用ExecutorService.submit方法实现的此方法返回一个Futurefuture.get()会让当前线程阻塞直到Future关联的任务执行完毕。示例public class ThreadPoolExecutorTest2 {public static void main(String[] args) throws ExecutionException, InterruptedException {//自定义包含策略ThreadPoolExecutor executor new ThreadPoolExecutor(1, 5, 60,TimeUnit.SECONDS, new LinkedBlockingQueue(5),new DemoThreadFactory(订单创建组), new ThreadPoolExecutor.AbortPolicy());Future future executor.submit(() - {System.out.println(start thread!);try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(end thread!);return success;});executor.shutdown();System.out.println(main get resultfuture.get());}}输出同上。方式4FutureTask方式1——作为Runnable传给Thread执行线程池的submit方法传入的Callable对象本质上也是包装成一个FutureTask来执行。代码中使用FutureTask实现的FutureTask实现了Runnable接口并且内部带返回值所以可以传递给Thread直接运行futureTask.get()会阻塞当前线程直到FutureTask构造方法传递的任务执行完毕get方法才会返回。示例public class FutureTaskTest {public static void main(String[] args) throws ExecutionException, InterruptedException {//创建一个FutureTaskFutureTask futureTask new FutureTask(() - {System.out.println(start thread!);try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(end thread!);return success;});//将futureTask传递一个线程运行new Thread(futureTask).start();//futureTask.get()会阻塞当前线程直到futureTask执行完毕String result futureTask.get();System.out.println(main get result result);}}方式5FutureTask方式2——构造FutureTask对象及执行内容直接在Thread里面跑run方法当futureTask的run()方法执行完毕之后futureTask.get()会从阻塞中返回。示例public class FutureTaskTest1 {public static void main(String[] args) throws ExecutionException, InterruptedException {//创建一个FutureTaskFutureTask futureTask new FutureTask(() - {System.out.println(start thread!);try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(end thread!);return success;});//将futureTask传递一个线程运行new Thread(()-futureTask.run()).start();//futureTask.get()会阻塞当前线程直到futureTask执行完毕String result futureTask.get();System.out.println(main get result result);}}方式6CompletableFuture方式实现CompletableFuture.supplyAsync可以用来异步执行一个带返回值的任务调用completableFuture.get()会阻塞当前线程直到任务执行完毕get方法才会返回。public class CompletableFutureTest4 {public static void main(String[] args) throws Exception {CompletableFuture future CompletableFuture.supplyAsync(() - {System.out.println(start thread!);try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(end thread!);return success;});// future.get()会阻塞当前线程直到获得结果System.out.println(main get resultfuture.get());}}高并发中计数器的四种实现方式需求一个jvm中实现一个计数器功能需保证多线程情况下数据正确性。我们来模拟50个线程每个线程对计数器递增100万次最终结果应该是5000万。我们使用4种方式实现看一下其性能然后引出为什么需要使用LongAdder、LongAccumulator。方式一使用加锁的方式实现——synchronized或Lock从示例输出结果看ReentrantLock的效率明显比synchronized差了2-3倍。示例public class SynchronizeCalculator {private static long count 0;private static Lock lock new ReentrantLock();public synchronized static void incrment() {count;}public static void incrmentByLock() {lock.lock();try {count;}finally {lock.unlock();}}public static void main(String[] args) throws InterruptedException {for (int i 0; i 5; i) {count 0;averageTest();}}public static void averageTest() throws InterruptedException {long t1 System.currentTimeMillis();//自定义包含策略ThreadPoolExecutor executor new ThreadPoolExecutor(50, 50, 60,TimeUnit.SECONDS, new LinkedBlockingQueue(5),new DemoThreadFactory(订单创建组), new ThreadPoolExecutor.AbortPolicy());CountDownLatch latch new CountDownLatch(50);for (int i 0; i 50; i) {executor.execute(() - {try {for (int j 0; j 1000000; j) {incrment();//incrmentByLock();}} finally {latch.countDown();}});}latch.await();long t2 System.currentTimeMillis();System.out.println(String.format(结果%s,耗时(ms)%s, count, (t2 - t1)));executor.shutdown();}}输出//synchronized结果50000000,耗时(ms)490结果50000000,耗时(ms)1574结果50000000,耗时(ms)399结果50000000,耗时(ms)395结果50000000,耗时(ms)396//lock结果50000000,耗时(ms)1289结果50000000,耗时(ms)1239结果50000000,耗时(ms)1224结果50000000,耗时(ms)1219结果50000000,耗时(ms)1246方式2AtomicLong实现AtomicLong内部采用CAS的方式实现并发量大的情况下CAS失败率比较高导致性能比synchronized还低一些。并发量不是太大的情况下CAS性能还是可以的。示例public class AtomicLongTest {private static AtomicLong count new AtomicLong(0);public static void main(String[] args) throws InterruptedException {for (int i 0; i 5; i) {count.set(0);averageTest();}}public static void averageTest() throws InterruptedException {long t1 System.currentTimeMillis();//自定义包含策略ThreadPoolExecutor executor new ThreadPoolExecutor(50, 50, 60,TimeUnit.SECONDS, new LinkedBlockingQueue(5),new DemoThreadFactory(订单创建组), new ThreadPoolExecutor.AbortPolicy());CountDownLatch latch new CountDownLatch(50);for (int i 0; i 50; i) {executor.execute(() - {try {for (int j 0; j 1000000; j) {count.getAndIncrement();}} finally {latch.countDown();}});}latch.await();long t2 System.currentTimeMillis();System.out.println(String.format(结果%s,耗时(ms)%s, count.get(), (t2 - t1)));executor.shutdown();}}输出结果50000000,耗时(ms)1018结果50000000,耗时(ms)1442结果50000000,耗时(ms)1033结果50000000,耗时(ms)935结果50000000,耗时(ms)1320方式3LongAdder实现——相当于锁分段技术先介绍一下LongAdder说到LongAdder不得不提的就是AtomicLongAtomicLong是JDK1.5开始出现的里面主要使用了一个long类型的value作为成员变量然后使用循环的CAS操作去操作value的值并发量比较大的情况下CAS操作失败的概率较高内部失败了会重试导致耗时可能会增加。LongAdder是JDK1.8开始出现的所提供的API基本上可以替换掉原先的AtomicLong。LongAdder在并发量比较大的情况下操作数据的时候相当于把这个数字分成了很多份数字然后交给多个人去管控每个管控者负责保证部分数字在多线程情况下操作的正确性。当多线程访问的时通过hash算法映射到具体管控者去操作数据最后再汇总所有的管控者的数据得到最终结果。相当于降低了并发情况下锁的粒度所以效率比较高看一下下面的图方便理解示例代码中new LongAdder()创建一个LongAdder对象内部数字初始值是0调用increment()方法可以对LongAdder内部的值原子递增1。reset()方法可以重置LongAdder的值使其归0。public class LongAdderTest {private static LongAdder count new LongAdder();public static void main(String[] args) throws InterruptedException {for (int i 0; i 5; i) {count.reset();averageTest();}}public static void averageTest() throws InterruptedException {long t1 System.currentTimeMillis();//自定义包含策略ThreadPoolExecutor executor new ThreadPoolExecutor(50, 50, 60,TimeUnit.SECONDS, new LinkedBlockingQueue(5),new DemoThreadFactory(订单创建组), new ThreadPoolExecutor.AbortPolicy());CountDownLatch latch new CountDownLatch(50);for (int i 0; i 50; i) {executor.execute(() - {try {for (int j 0; j 1000000; j) {count.increment();}} finally {latch.countDown();}});}latch.await();long t2 System.currentTimeMillis();System.out.println(String.format(结果%s,耗时(ms)%s, count.sum(), (t2 - t1)));executor.shutdown();}}输出结果50000000,耗时(ms)209结果50000000,耗时(ms)133结果50000000,耗时(ms)149结果50000000,耗时(ms)146结果50000000,耗时(ms)148方式4LongAccumulator实现LongAccumulator介绍LongAccumulator是LongAdder的功能增强版。LongAdder的API只有对数值的加减而LongAccumulator提供了自定义的函数操作其构造函数如下/*** accumulatorFunction需要执行的二元函数(接收2个long作为形参并返回1个long)* identity初始值**/public LongAccumulator(LongBinaryOperator accumulatorFunction, long identity) {this.function accumulatorFunction;base this.identity identity;}示例LongAccumulator的效率和LongAdder差不多不过更灵活一些。调用new LongAdder()等价于new LongAccumulator((x, y) - x y, 0L)。从上面4个示例的结果来看LongAdder、LongAccumulator全面超越同步锁及AtomicLong的方式建议在使用AtomicLong的地方可以直接替换为LongAdder、LongAccumulator吞吐量更高一些。public class LongAccumulatorTest {private static volatile LongAccumulator count new LongAccumulator((x, y) - x y, 0);public static void main(String[] args) throws InterruptedException {for (int i 0; i 5; i) {count.reset();averageTest();}}public static void averageTest() throws InterruptedException {long t1 System.currentTimeMillis();//自定义包含策略ThreadPoolExecutor executor new ThreadPoolExecutor(50, 50, 60,TimeUnit.SECONDS, new LinkedBlockingQueue(5),new DemoThreadFactory(订单创建组), new ThreadPoolExecutor.AbortPolicy());CountDownLatch latch new CountDownLatch(50);for (int i 0; i 50; i) {executor.execute(() - {try {for (int j 0; j 1000000; j) {count.accumulate(1);}} finally {latch.countDown();}});}latch.await();long t2 System.currentTimeMillis();System.out.println(String.format(结果%s,耗时(ms)%s, count.longValue(), (t2 - t1)));executor.shutdown();}}输出结果50000000,耗时(ms)152结果50000000,耗时(ms)148结果50000000,耗时(ms)137结果50000000,耗时(ms)137结果50000000,耗时(ms)144疑问Q:LongAccumulator.reset方法并不能重置重置LongAccumulator的identity初始值正确使其恢复原来的初始值。当初始值为0是不会发生这个问题而当我们设置初始值如1时就会导致后续的计算操作增加了5份初始值目前猜测原因是因为代码中LongAccumulator在并发量比较大的情况下操作数据的时候相当于把这个数字分成了很多份数字 而初始化的时候也是初始化了多份数据导致初始值叠加了多份。不知道这是个bug么待解惑。在此记录下来希望有遇到这种情况的同学注意。解决方法便是要么初始值identity0不会有这种问题或者有需要使用reset方法重置的改为重新创建个LongAccumulator处理。源码public void reset() {Cell[] as cells; Cell a;base identity;if (as ! null) {for (int i 0; i as.length; i) {if ((a as[i]) ! null)//对多个cell进行初始值赋值导致后面计算叠加了多份初始值a.value identity;}}}示例public class LongAccumulatorTest {//设置初始值为1查看输出结果private static volatile LongAccumulator count new LongAccumulator((x, y) - x y, 1);public static void main(String[] args) throws InterruptedException {for (int i 0; i 5; i) {count.reset();averageTest();}}public static void averageTest() throws InterruptedException {long t1 System.currentTimeMillis();//自定义包含策略ThreadPoolExecutor executor new ThreadPoolExecutor(50, 50, 60,TimeUnit.SECONDS, new LinkedBlockingQueue(5),new DemoThreadFactory(订单创建组), new ThreadPoolExecutor.AbortPolicy());CountDownLatch latch new CountDownLatch(50);for (int i 0; i 50; i) {executor.execute(() - {try {for (int j 0; j 1000000; j) {count.accumulate(1);}} finally {latch.countDown();}});}latch.await();long t2 System.currentTimeMillis();System.out.println(String.format(结果%s,耗时(ms)%s, count.longValue(), (t2 - t1)));executor.shutdown();}}输出这时候你会发现只有第一次计算是正确的只要使用了rest方法重置就会导致这个错误。结果50000001,耗时(ms)185结果50000005,耗时(ms)143结果50000005,耗时(ms)139结果50000005,耗时(ms)162结果50000005,耗时(ms)142演示公平锁和非公平锁先理解一下什么是公平锁、非公平锁公平锁和非公平锁体现在别人释放锁的一瞬间如果前面已经有排队的新来的是否可以插队如果可以插队表示是非公平的如果不可用插队只能排在最后面是公平的方式。示例Slf4jpublic class ReentrantLockTest2 {public static void main(String[] args) throws InterruptedException {ReentrantLockTest2.LockTest(false);TimeUnit.SECONDS.sleep(4);log.error(-------------------------------);ReentrantLockTest2.LockTest(true);}public static void LockTest(boolean fair) throws InterruptedException {ReentrantLock lock new ReentrantLock(fair);new Thread(() - {lock.lock();try {log.error(Thread.currentThread().getName() start!);TimeUnit.SECONDS.sleep(3);new Thread(() - {//注意线程池要当前线程创建的才能使用如果传给新开的线程会获取不到线程池test(后到组,lock);}).start();//test(executorAfter,lock);log.error(Thread.currentThread().getName() end!);} catch (InterruptedException e) {e.printStackTrace();} finally {lock.unlock();}}, Hold Lock 4 Test Thread).start();test(先到组,lock);TimeUnit.SECONDS.sleep(3);}private static void test(String name,Lock lock){//自定义包含策略ThreadPoolExecutor executor new ThreadPoolExecutor(10, 10, 60,TimeUnit.SECONDS, new LinkedBlockingQueue(5),new DemoThreadFactory(name), new ThreadPoolExecutor.AbortPolicy());for (int i 0; i 10; i) {executor.execute(() - {lock.lock();try {log.error(获取到锁!);} finally {lock.unlock();}});}executor.shutdown();}}输出14:45:23.204 [Hold Lock 4 Test Thread] ERROR com.self.current.ReentrantLockTest2 - Hold Lock 4 Test Thread start!14:45:26.211 [Hold Lock 4 Test Thread] ERROR com.self.current.ReentrantLockTest2 - Hold Lock 4 Test Threadend!14:45:26.211 [From DemoThreadFactorys 先到组-Worker-1] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.211 [From DemoThreadFactorys 先到组-Worker-2] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.212 [From DemoThreadFactorys 先到组-Worker-3] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.212 [From DemoThreadFactorys 先到组-Worker-4] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.212 [From DemoThreadFactorys 先到组-Worker-5] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.212 [From DemoThreadFactorys 先到组-Worker-6] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.212 [From DemoThreadFactorys 先到组-Worker-7] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.212 [From DemoThreadFactorys 先到组-Worker-8] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.212 [From DemoThreadFactorys 后到组-Worker-4] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.212 [From DemoThreadFactorys 先到组-Worker-9] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.213 [From DemoThreadFactorys 后到组-Worker-8] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.218 [From DemoThreadFactorys 后到组-Worker-10] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.218 [From DemoThreadFactorys 先到组-Worker-10] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.218 [From DemoThreadFactorys 后到组-Worker-2] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.218 [From DemoThreadFactorys 后到组-Worker-1] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.219 [From DemoThreadFactorys 后到组-Worker-3] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.219 [From DemoThreadFactorys 后到组-Worker-5] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.219 [From DemoThreadFactorys 后到组-Worker-6] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.219 [From DemoThreadFactorys 后到组-Worker-7] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:26.219 [From DemoThreadFactorys 后到组-Worker-9] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:30.205 [main] ERROR com.self.current.ReentrantLockTest2 - -------------------------------14:45:30.205 [Hold Lock 4 Test Thread] ERROR com.self.current.ReentrantLockTest2 - Hold Lock 4 Test Thread start!14:45:33.206 [Hold Lock 4 Test Thread] ERROR com.self.current.ReentrantLockTest2 - Hold Lock 4 Test Threadend!14:45:33.206 [From DemoThreadFactorys 先到组-Worker-1] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.206 [From DemoThreadFactorys 先到组-Worker-2] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.209 [From DemoThreadFactorys 先到组-Worker-3] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.209 [From DemoThreadFactorys 先到组-Worker-4] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.209 [From DemoThreadFactorys 先到组-Worker-5] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.209 [From DemoThreadFactorys 先到组-Worker-6] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.210 [From DemoThreadFactorys 先到组-Worker-7] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.210 [From DemoThreadFactorys 先到组-Worker-8] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.210 [From DemoThreadFactorys 先到组-Worker-9] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.210 [From DemoThreadFactorys 先到组-Worker-10] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.210 [From DemoThreadFactorys 后到组-Worker-2] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.210 [From DemoThreadFactorys 后到组-Worker-1] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.211 [From DemoThreadFactorys 后到组-Worker-6] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.211 [From DemoThreadFactorys 后到组-Worker-7] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.211 [From DemoThreadFactorys 后到组-Worker-5] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.211 [From DemoThreadFactorys 后到组-Worker-4] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.211 [From DemoThreadFactorys 后到组-Worker-3] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.211 [From DemoThreadFactorys 后到组-Worker-9] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.212 [From DemoThreadFactorys 后到组-Worker-10] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!14:45:33.212 [From DemoThreadFactorys 后到组-Worker-8] ERROR com.self.current.ReentrantLockTest2 - 获取到锁!google提供的一些好用的并发工具类关于并发方面的juc已帮我们提供了很多好用的工具而谷歌在此基础上做了扩展使并发编程更容易这些工具放在guava.jar包中。guava maven配置com.google.guavaguava27.0-jreguava中常用几个类MoreExecutors提供了一些静态方法是对juc中的Executors类的一个扩展。Futures也提供了很多静态方法是对juc中Future的一个扩展。案例1异步执行任务完毕之后回调——相当于CompletableFuture的whenComplete方法ListeningExecutorService接口继承于juc中的ExecutorService接口对ExecutorService做了一些扩展看其名字中带有Listening说明这个接口自带监听的功能可以监听异步执行任务的结果。通过MoreExecutors.listeningDecorator创建一个ListeningExecutorService对象需传递一个ExecutorService参数传递的ExecutorService负责异步执行任务。ListeningExecutorService的submit方法用来异步执行一个任务返回ListenableFutureListenableFuture接口继承于juc中的Future接口对Future做了扩展使其带有监听的功能。调用submit.addListener可以在执行的任务上添加监听器当任务执行完毕之后会回调这个监听器中的方法。ListenableFuture的get方法会阻塞当前线程直到任务执行完毕。另一种回调方式是通过调用Futures的静态方法addCallback在异步执行的任务中添加回调回调的对象是一个FutureCallback此对象有2个方法任务执行成功调用onSuccess执行失败调用onFailure。失败的情况可以将代码中int i 10 / 0;注释去掉执行一下可以看看效果。示例Slf4jpublic class GuavaTest {//相当于CompletableFuture的whenComplete方法public static void main1(String[] args) throws ExecutionException, InterruptedException {//创建一个线程池ExecutorService delegate Executors.newFixedThreadPool(5);try {ListeningExecutorService executorService MoreExecutors.listeningDecorator(delegate);//异步执行一个任务ListenableFuture submit executorService.submit(() - {log.error({}, System.currentTimeMillis());//休眠2秒默认耗时TimeUnit.SECONDS.sleep(2);log.error({}, System.currentTimeMillis());return 10;});//当任务执行完毕之后回调对应的方法submit.addListener(() - {log.error(任务执行完毕了我被回调了);}, MoreExecutors.directExecutor());log.error({}, submit.get());} finally {delegate.shutdown();}}//相当于CompletableFuture的whenComplete方法public static void main(String[] args) throws ExecutionException, InterruptedException {ThreadPoolExecutor executor new ThreadPoolExecutor(5, 10, 60,TimeUnit.SECONDS, new LinkedBlockingQueue(10),new DemoThreadFactory(订单创建组), new ThreadPoolExecutor.AbortPolicy());ListeningExecutorService service MoreExecutors.listeningDecorator(executor);try {ListenableFuture future service.submit(() - {log.error({}, System.currentTimeMillis());//休眠2秒默认耗时TimeUnit.SECONDS.sleep(2);//int i 10 / 0;log.error({}, System.currentTimeMillis());return 10;});Futures.addCallback(future, new FutureCallback() {Overridepublic void onSuccess(Integer integer) {log.error(执行成功:{}, integer);}Overridepublic void onFailure(Throwable throwable) {log.error(执行失败:{}, throwable.getMessage());}});log.error({}, future.get());} finally {service.shutdown();}}}输出15:32:54.480 [From DemoThreadFactorys 订单创建组-Worker-1] ERROR com.self.current.GuavaTest - 159980957447715:32:56.487 [From DemoThreadFactorys 订单创建组-Worker-1] ERROR com.self.current.GuavaTest - 159980957648715:32:56.488 [main] ERROR com.self.current.GuavaTest - 1015:32:56.488 [From DemoThreadFactorys 订单创建组-Worker-1] ERROR com.self.current.GuavaTest - 执行成功:10示例2获取一批异步任务的执行结果——Futures.allAsList(futureList).get()结果中按顺序输出了6个异步任务的结果此处用到了Futures.allAsList方法看一下此方法的声明public static ListenableFuture allAsList(Iterable extends ListenableFuture extends V futures)传递一批ListenableFuture返回一个ListenableFuture内部将一批结果转换为了一个ListenableFuture对象。示例public static void main(String[] args) throws ExecutionException, InterruptedException {log.error(star);ExecutorService delegate Executors.newFixedThreadPool(5);try {ListeningExecutorService executorService MoreExecutors.listeningDecorator(delegate);List futureList new ArrayList();for (int i 5; i 0; i--) {int j i;futureList.add(executorService.submit(() - {TimeUnit.SECONDS.sleep(j);return j;}));}//把多个ListenableFuture转换为一个ListenableFuture//ListenableFuture listListenableFuture Futures.allAsList(futureList);//获取一批任务的执行结果List resultList Futures.allAsList(futureList).get();//输出resultList.forEach(item - {log.error({}, item);});} finally {delegate.shutdown();}}输出15:45:51.160 [main] ERROR com.self.current.GuavaTest - star15:45:56.185 [main] ERROR com.self.current.GuavaTest - 515:45:56.185 [main] ERROR com.self.current.GuavaTest - 415:45:56.185 [main] ERROR com.self.current.GuavaTest - 315:45:56.185 [main] ERROR com.self.current.GuavaTest - 215:45:56.185 [main] ERROR com.self.current.GuavaTest - 115:45:56.185 [main] ERROR com.self.current.GuavaTest - 0示例3一批任务异步执行完毕之后回调——包装futureList传递给Futures.addCallback 方法异步执行一批任务最后计算其和,代码中异步执行了一批任务所有任务完成之后回调了上面的onSuccess方法内部对所有的结果进行sum操作。示例public static void main(String[] args) throws ExecutionException, InterruptedException {log.error(start);ExecutorService delegate Executors.newFixedThreadPool(5);try {ListeningExecutorService executorService MoreExecutors.listeningDecorator(delegate);List futureList new ArrayList();for (int i 5; i 0; i--) {int j i;futureList.add(executorService.submit(() - {TimeUnit.SECONDS.sleep(j);return j;}));}//把多个ListenableFuture转换为一个ListenableFutureListenableFuture listListenableFuture Futures.allAsList(futureList);Futures.addCallback(listListenableFuture, new FutureCallback() {Overridepublic void onSuccess(List result) {log.error(result中所有结果之和result.stream().reduce(Integer::sum).get());}Overridepublic void onFailure(Throwable throwable) {log.error(执行任务发生异常: throwable.getMessage(), throwable);}});} finally {delegate.shutdown();}}输出15:57:00.539 [main] ERROR com.self.current.GuavaTest - start15:57:05.560 [pool-2-thread-1] ERROR com.self.current.GuavaTest - result中所有结果之和15其他疑问Q:运行下面这个例子结束不了debug倒是可以这是为什么呢Thread[Monitor Ctrl-Break,5,main]是哪来的public class VolatileTest1 {public static volatile int num 0;public static void add(){num;}public static void main(String[] args) throws InterruptedException {Thread[] threads new Thread[10];for (Thread thread : threads) {thread new Thread(()-{for (int i 0; i 1000; i) {VolatileTest1.add();}});thread.start();thread.join();}//2//java.lang.ThreadGroup[namemain,maxpri10]// Thread[main,5,main]// Thread[Monitor Ctrl-Break,5,main]//结束不了debug倒是可以这是为什么呢Thread[Monitor Ctrl-Break,5,main]是哪来的while (Thread.activeCount() 1){Thread.yield();System.out.println(Thread.activeCount());ThreadGroup parent Thread.currentThread().getThreadGroup();parent.list();}System.out.println(numnum);}}
