• Java 多线程的返回对象和资源独享线程

    1. 多线程的返回对象-Future

    1.1 Future 

    如果你在创建线程时，使用的是 Runnable 接口，那么此时你是无法获取线程执行结果的，如果想要获取线程的执行结果，需要实现 Callable 接口，示例如下：

    1. public class J0_Callable {
    2.  
    3.     static class Task implements Callable {
    4.  
    5.         @Override
    6.         public Integer call() throws InterruptedException {
    7.             Thread.sleep(3000);
    8.             return 100;
    9.         }
    10.     }
    11.  
    12.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    13.         ExecutorService executors = Executors.newSingleThreadExecutor();
    14.         Future submit = executors.submit(new Task());
    15.         System.out.println("计算结果为:" + submit.get());
    16.         executors.shutdown();
    17.     }
    18. }

    此时通过 ExecutorService.submit() 进行提交，得到的是一个 Future 对象，它包含了线程的执行结果，当你调用其 get() 方法时，它会阻塞直至获取到线程的返回结果。

    1.2  FutureTask

    使用 Callable 接口的限制是：其只能使用线程池提交，而不能使用单独的线程进行提交。如果想要使用单独的线程提交，可以使用 FutureTask 对其进行包装，FutureTask 是 Runnable 接口的实现类，可以用于任何场景下的提交，示例如下：

    1. static class Task implements Callable<Integer> {
    2.     @Override
    3.     public Integer call() throws InterruptedException {
    4.         Thread.sleep(3000);
    5.         return 100;
    6.     }
    7. }
    8.  
    9. public static void main(String[] args) throws ExecutionException, InterruptedException {
    10.     FutureTask<Integer> futureTask01 = new FutureTask<>(new Task());
    11.     FutureTask<Integer> futureTask02 = new FutureTask<>(new Task());
    12.     // 使用独立的线程执行
    13.     new Thread(futureTask01).start();
    14.     ExecutorService executorService = Executors.newSingleThreadExecutor();
    15.     // 使用线程池提交
    16.     executorService.submit(futureTask02);
    17.     System.out.println("futureTask01 计算结果为：" + futureTask01.get());
    18.     System.out.println("futureTask02 计算结果为：" + futureTask01.get());
    19.     executorService.shutdown();
    20. }

    1.3  CompletableFuture

    CompletableFuture 是 JDK 8 提供的增强后 Future ，它支持流式调用，等待唤醒等一系列新的功能：

    1.3.1 等待唤醒

    1. public class J2_CompletableFuture {
    2.  
    3.     static class Compute implements Runnable {
    4.  
    5.         private CompletableFuture<Integer> future;
    6.  
    7.         Compute(CompletableFuture<Integer> future) {
    8.             this.future = future;
    9.         }
    10.  
    11.         @Override
    12.         public void run() {
    13.             try {
    14.                 System.out.println("子线程等待主线程运算完成····");
    15.                 Integer integer = future.get();
    16.                 System.out.println("子线程完成后续运算:" + integer * integer);
    17.             } catch (InterruptedException | ExecutionException e) {
    18.                 e.printStackTrace();
    19.             }
    20.         }
    21.     }
    22.  
    23.     public static void main(String[] args) throws InterruptedException {
    24.         int intermediateResult;
    25.         CompletableFuture<Integer> future = new CompletableFuture<>();
    26.         // 启动子线程
    27.         new Thread(new Compute(future)).start();
    28.         System.out.println("启动主线程");
    29.         Thread.sleep(2000);
    30.         System.out.println("主线程计算完成");
    31.         // 假设主线程计算结果为 100
    32.         intermediateResult = 100;
    33.         // 传递主线程的计算结果给子线程
    34.         future.complete(intermediateResult);
    35.     }
    36. }
    37.  
    38. // 输出
    39. 启动主线程
    40.     子线程等待主线程运算完成····
    41.     主线程计算完成
    42.     子线程完成后续运算:10000

    1.3.2 supplyAsync

    CompletableFuture 的 supplyAsync 可以将一个正常的方法以异步的方式来执行：

    1. public class J3_SupplyAsync {
    2.  
    3.     private static Integer compute() {
    4.         try {
    5.             Thread.sleep(2000);
    6.         } catch (InterruptedException e) {
    7.             e.printStackTrace();
    8.         }
    9.         System.out.println("子线程计算完成");
    10.         return 100;
    11.     }
    12.  
    13.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    14.         CompletableFuture<Integer> supplyAsync = CompletableFuture.supplyAsync(J3_SupplyAsync::compute);
    15.         System.out.println("主线程等待子线程计算完成");
    16.         Integer integer = supplyAsync.get();
    17.         System.out.println("主线程计算完成:" + integer * integer);
    18.     }
    19. }

    1.3.3 流式调用

    CompletableFuture 支持大部分流式处理的特性，示例如下：

    1. public class J4_StreamingCall {
    2.  
    3.     private static Integer compute() {
    4.         System.out.println("compute所在线程：" + Thread.currentThread().getId());
    5.         try {
    6.             Thread.sleep(1000);
    7.         } catch (InterruptedException e) {
    8.             e.printStackTrace();
    9.         }
    10.         return 100;
    11.     }
    12.  
    13.     private static Integer multi(Integer integer) {
    14.         try {
    15.             System.out.println("multi所在线程：" + Thread.currentThread().getId());
    16.             Thread.sleep(1000);
    17.         } catch (InterruptedException e) {
    18.             e.printStackTrace();
    19.         }
    20.         return integer * integer;
    21.     }
    22.  
    23.     private static void accept(Integer integer) {
    24.         System.out.println("accept所在线程：" + Thread.currentThread().getId());
    25.         System.out.println("accept方法消费掉计算结果:" + integer);
    26.     }
    27.  
    28.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    29.         CompletableFuture<Void> future = CompletableFuture.supplyAsync(J4_StreamingCall::compute)
    30.             .thenApply(J4_StreamingCall::multi)
    31.             .thenAccept(J4_StreamingCall::accept)   //值在这一步被消费掉了
    32.             .thenAccept(x -> System.out.println("运算结果:" + x));
    33.         future.get(); //类似于流式计算的惰性求值，如果缺少这一步，不会有任何输出
    34.     }
    35. }

    1.3.4 组合多个 CompletableFuture

    除了使用单个的 CompletableFuture，还可以通过 thenCompose 或 thenCombineAsync 来组合多个 CompletableFuture：

    1. public class J6_Combination {
    2.  
    3.     private static Integer compute() {
    4.         System.out.println("compute 所在线程：" + Thread.currentThread().getId());
    5.         return 100;
    6.     }
    7.  
    8.     private static Integer multi(Integer integer) {
    9.         System.out.println("epr 所在线程：" + Thread.currentThread().getId());
    10.         return integer * integer;
    11.     }
    12.  
    13.  
    14.     public static void main(String[] args) throws ExecutionException, InterruptedException {
    15.         // 组合实现方式1 thenCompose 一个计算的输入依赖另外一个计算的结果
    16.         CompletableFuture<Void> future01 = CompletableFuture.supplyAsync(J6_Combination::compute)
    17.             .thenCompose(x -> CompletableFuture.supplyAsync(() -> multi(x)))
    18.             .thenAccept(x -> System.out.println("运算结果:" + x));    // 运算结果:10000
    19.         future01.get();
    20.  
    21.         System.out.println();
    22.  
    23.         // 组合实现方式2 thenCombineAsync 两个计算之间不依赖
    24.         CompletableFuture<Integer> future02 = CompletableFuture.supplyAsync(J6_Combination::compute);
    25.         CompletableFuture<Integer> future03 = CompletableFuture.supplyAsync(() -> J6_Combination.multi(100));
    26.         CompletableFuture<Integer> futureAll = future02.thenCombineAsync(future03, (x, y) -> x + y);
    27.         System.out.println("运算结果:" + futureAll.get()); // 运算结果:10100
    28.  
    29.     }
    30. }

    2.资源独享线程-ThreadLocal

    ThreadLocal 是以增加资源的方式来避免竞态，它会为每一个线程创建一份私有的资源，从而避免对公共资源的竞争。实例如下：

    1. /**
    2.  * 线程不安全的SimpleDateFormat
    3.  */
    4. public class J1_ThreadUnsafe {
    5.  
    6.     private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    7.     private static int sum = 1000;
    8.     private static CountDownLatch countDownLatch = new CountDownLatch(sum);
    9.     private static AtomicInteger atomicInteger = new AtomicInteger(0);
    10.  
    11.     static class Task implements Runnable {
    12.  
    13.         @Override
    14.         public void run() {
    15.             try {
    16.                 Date parse = sdf.parse("2018-08-08 08:08:08");
    17.                 System.out.println(parse);
    18.                 atomicInteger.incrementAndGet();
    19.             } catch (ParseException e) {
    20.                 e.printStackTrace();
    21.             } finally {
    22.                 countDownLatch.countDown();
    23.             }
    24.         }
    25.     }
    26.  
    27.     public static void main(String[] args) throws InterruptedException {
    28.         ExecutorService executorService = Executors.newFixedThreadPool(10);
    29.         for (int i = 0; i < sum; i++) {
    30.             executorService.execute(new Task());
    31.         }
    32.         countDownLatch.await();
    33.         System.out.println("格式化成功次数为：" + atomicInteger.get());
    34.     }
    35. }

    因为 SimpleDateFormat 是线程不安全的，因此其格式化成功的次数总是小于 100 次，此时可以使用 ThreadLocal 进行改写，让每个线程都持有自己独立的格式化器，具体如下：

    1. public class J2_ThreadSafe {
    2.  
    3.     private static ThreadLocal threadLocal = new ThreadLocal<>();
    4.     private static int sum = 1000;
    5.     private static CountDownLatch countDownLatch = new CountDownLatch(sum);
    6.     private static AtomicInteger atomicInteger = new AtomicInteger(0);
    7.  
    8.     static class Task implements Runnable {
    9.  
    10.         @Override
    11.         public void run() {
    12.             try {
    13.                 // 如果当前线程中不存在该值，则创建一个
    14.                 if (threadLocal.get() == null) {
    15.                     threadLocal.set(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
    16.                 }
    17.                 // 使用线程私有的SimpleDateFormat
    18.                 Date parse = threadLocal.get().parse("2018-08-08 08:08:08");
    19.                 System.out.println(parse);
    20.                 atomicInteger.incrementAndGet();
    21.             } catch (ParseException e) {
    22.                 e.printStackTrace();
    23.             } finally {
    24.                 countDownLatch.countDown();
    25.             }
    26.         }
    27.     }
    28.  
    29.     public static void main(String[] args) throws InterruptedException {
    30.         ExecutorService executorService = Executors.newFixedThreadPool(10);
    31.         for (int i = 0; i < sum; i++) {
    32.             executorService.execute(new Task());
    33.         }
    34.         countDownLatch.await();
    35.         System.out.println("格式化成功次数为：" + atomicInteger.get());
    36.         executorService.shutdown();
    37.     }
    38. }
  • 相关阅读:
    6、ES单机设置用户名密码、集群设置用户名密码、es-head登录、如何去掉密码
    “如何实现高效的应用交付”鲁班会开发者训练营厦门站进行时
    84.（cesium之家）cesium模型在地形上运动
    过五关，斩六将！「网易/美团/菜鸟」已拿offer【Java岗】
    如何为开源项目和社区做贡献 -- 你应该知道的十件事
    Java 的基本类型与引用类型的深拷贝和浅拷贝
    uniapp项目实践总结(十三)封装文件操作方法
    这可能是目前最全的Redis高可用技术解决方案
    mysql之搭建MHA架构实现高可用
    Vue3中全局配置 axios 的两种方式
  • 原文地址：https://blog.csdn.net/shangjg03/article/details/134229437