Java并發JUC（java.util.concurrent）ForkJoin/異步回呼

ForkJoin

• 什么是 ForkJoin

  • ForkJoin 在 JDK 1.7 ， 并行執行任務！提高效率，大資料量！

  

• ForkJoin處理流程：作業竊取
  
  

• 試驗代碼：

  • MyForkJoinTask:

    package icu.lookyousmileface.forkjoin;
    
    import java.util.concurrent.RecursiveTask;
    
    /**
     * @author starrysky
     * @title: MyForkJoinTask
     * @projectName Juc_Pro
     * @description: ForkJon,必須要繼承RecursiceTask
     *  * 求和計算的任務！
     *  * 3000   6000（ForkJoin）  9000（Stream并行流）
     *  * // 如何使用 forkjoin
     *  * // 1、forkjoinPool 通過它來執行
     *  * // 2、計算任務 forkjoinPool.execute(ForkJoinTask task)
     *  * // 3. 計算類要繼承 ForkJoinTask
     * @date 2021/1/301:11 上午
     */
    class MyForkJoinTask extends RecursiveTask<Long> {
        //開始和結束位置數
        private Long start;
        private Long end;
        //臨界值
        private Long temp = 10000L;
    
        public MyForkJoinTask(Long start, Long end) {
            this.start = start;
            this.end = end;
        }
        //計算方法
        @Override
        protected Long compute() {
            //小于臨界值就進行計算不拆分
            if ((end-start)<temp){
                Long sum = 0L;
                for (Long i = start; i <= end; i++) {
                    sum += i;
                }
                return sum;
            }else {
                //取中位數
                Long mdie = (start+end)/2;
                MyForkJoinTask task1 = new MyForkJoinTask(start, mdie);
                //拆分任務，把任務壓入執行緒佇列
                task1.fork();
                MyForkJoinTask task2 = new MyForkJoinTask(mdie + 1, end);
                //拆分任務，把任務壓入執行緒佇列
                task2.fork();
                //結果匯聚
                return task1.join()+task2.join();
            }
        }
    }
    

  • MainTask:

    package icu.lookyousmileface.forkjoin;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    import java.util.stream.LongStream;
    
    /**
     * @author starrysky
     * @title: MainTask
     * @projectName Juc_Pro
     * @description: ForkJoin主任務
     * @date 2021/1/301:31 上午
     */
    public class MainTask {
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            /**
             * 使用ForkJoin，適合大資料量
             */
            //創建forkjoin池
    //        ForkJoinPool forkJoinPool = new ForkJoinPool();
    //        //創建自己的ForkJoin計算程式
    //        ForkJoinTask forkJoinTask = new MyForkJoinTask(0L, 10_0000_0000L);
    //        //提交計算任務
    //        ForkJoinTask<Long> submit = forkJoinPool.submit(forkJoinTask);
    //        //獲得計算的結果
    //        Long aLong = submit.get();
    //        System.out.println(aLong);
    
            /**
             * 使用stream并行流,非常快
             */
            long result = LongStream.rangeClosed(0L, 10_0000_0000L).parallel().reduce(0, Long::sum);
            System.out.println(result);
    
        }
    }
    

異步回呼

• 試驗代碼：

  package icu.lookyousmileface.completables;
  
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.TimeUnit;
  
  /**
   * @author starrysky
   * @title: CompletableUse
   * @projectName Juc_Pro
   * @description: CompletableFuture
   *  * 異步呼叫： CompletableFuture
   *  * // 異步執行
   *  * // 成功回呼
   *  * // 失敗回呼
   * @date 2021/1/302:13 上午
   */
  public class CompletableUse {
      public static void main(String[] args) throws ExecutionException, InterruptedException {
  //        //沒有回傳值的異步回呼
  //        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(()->{
  //            try {
  //                TimeUnit.SECONDS.sleep(3);
  //            } catch (InterruptedException e) {
  //                e.printStackTrace();
  //            }
  //            System.out.println(" 異步任務執行成功！");
  //        });
  //        System.out.println("main主執行緒");
  //        //獲取異步執行的結果
  //        completableFuture.get();
  
          //又回傳值的異步回呼
          CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
              System.out.println(Thread.currentThread().getName()+":supplyAsync=>ok");
              int sum = 10/0;
              return 1024;
          });
          //編譯
          //編譯成功
          System.out.println(completableFuture.whenComplete((u1,u2)->{
              System.out.println("t=>"+u1);//正常的回傳結果
              System.out.println("u=>"+u2);//錯誤資訊
              //編譯失敗
          }).exceptionally((e)->{
              e.printStackTrace();
              return 2233; //錯誤的回傳結果
          }).get());
      }
  }