并发编程JUC（中）

谁践踏了优雅 2023-10-02 10:29 31阅读 0赞

## 9、读写锁 ##

> 一个 `ReadWriteLock`保持一对一联系 `locks`，一个用于只读操作，一个用于写入，读操作可以多个线程而写操作只能一个线程

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16]

> 代码案例：做一个我们自己的cache缓存。分别有写入操作、读取操作，采用五个线程去写入，使用五个线程去读取，查看结果

代码示例：

/**
     * @author cVzhanshi
     * @create 2021-09-22 14:50
     */
    /**
     * ReadWriteLock
     */
    public class ReadWriteLockDemo {
        
        public static void main(String[] args) {
        
            MyCacheLock myCache = new MyCacheLock();
    
            //写入
            for (int i = 1; i <= 5; i++) {
        
                final int temp = i;
                new Thread(() -> {
        
                    myCache.put(temp +"",temp + "");
                },String.valueOf(i)).start();
            }
    
            //读取
            for (int i = 1; i <= 5; i++) {
        
                final int temp = i;
                new Thread(() -> {
        
                    myCache.get(temp + "");
                },String.valueOf(i)).start();
            }
        }
    }
    
    class MyCacheLock{
        
        private volatile Map<String,String> map=new HashMap<>();
        //使用读写锁
        private ReadWriteLock readWriteLock=new ReentrantReadWriteLock();
        //普通锁
        private Lock lock=new ReentrantLock();
    
        public void put(String key,String value){
        
            readWriteLock.writeLock().lock();
            try {
        
                //写入
                System.out.println(Thread.currentThread().getName()+" 线程 开始写入");
                map.put(key, value);
                System.out.println(Thread.currentThread().getName()+" 线程 写入OK");
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                readWriteLock.writeLock().unlock();
            }
    
        }
    
        public void get(String key) {
        
            readWriteLock.readLock().lock();
            try {
        
                //得到
                System.out.println(Thread.currentThread().getName() + " 线程 开始读取");
                String o = map.get(key);
                System.out.println(Thread.currentThread().getName() + " 线程 读取OK");
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                readWriteLock.readLock().unlock();
            }
        }
    
    }
    
    /**
     * 自定义缓存
     */
    class MyCache{
        
        private volatile Map<String,String> map=new HashMap<>();
        public void put(String key,String value){
        
            //写入
            System.out.println(Thread.currentThread().getName()+" 线程 开始写入");
            map.put(key, value);
            System.out.println(Thread.currentThread().getName()+" 线程 写入OK");
        }
    
        public String get(String key){
        
            //读取
            System.out.println(Thread.currentThread().getName()+" 线程 开始读取");
            String o = map.get(key);
            System.out.println(Thread.currentThread().getName()+" 线程 读取OK");
            return o;
        }
    }

*  未加锁情况下
    
    ![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 1]
 *  加锁情况下
    
    ![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 2]

> 小结

**ReadWriteLock**

*  读-读 可以共存
 *  读-写 不能共存
 *  写-写 不能共存

独占锁 ----> 写锁 -----> 一次只能被一个线程占用

共享锁 ----> 读锁 -----> 多个线程可以同时占有

## 10、阻塞队列 ##

### 10.1 简述 ###

对于队列与阻塞的理解

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 3]

阻塞队列：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 4]

**BlockingQueue**不是新的东西，blockingQueue 是Collection的一个子类。

**Collection与BlockingQueue的结构图**：

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 5]

**什么情况下会用到BlockingQueue**

*  多线程并发处理、线程池

### 10.2 四组API ###

**四组API**

<table> 
 <thead> 
  <tr> 
   <th>方式</th> 
   <th>抛出异常</th> 
   <th>不会抛出异常，有返回值</th> 
   <th>阻塞 等待</th> 
   <th>超时 等待</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>添加</td> 
   <td>add()</td> 
   <td>offer()</td> 
   <td>put()</td> 
   <td>offer(timenum，timeUnit)</td> 
  </tr> 
  <tr> 
   <td>移除</td> 
   <td>remove()</td> 
   <td>poll()</td> 
   <td>take()</td> 
   <td>poll(timenum，timeUnit)</td> 
  </tr> 
  <tr> 
   <td>判断队列首</td> 
   <td>element()</td> 
   <td>peek()</td> 
   <td>-</td> 
   <td>-</td> 
  </tr> 
 </tbody> 
</table>

**代码测试**

*  抛出异常

/**
     * 抛出异常
     */
    @Test
    public void test01(){
        
        //需要初始化队列的大小
        ArrayBlockingQueue queue = new ArrayBlockingQueue<>(3);
        System.out.println(queue.add("cvzhanshi1"));
        System.out.println(queue.add("cvzhanshi2"));
        System.out.println(queue.add("cvzhanshi3"));
        //抛出异常：java.lang.IllegalStateException: Queue full
        //System.out.println(queue.add("a"));
        System.out.println(queue.remove());
        System.out.println(queue.remove());
        System.out.println(queue.remove());
        //如果多移除一个
        //这也会造成 java.util.NoSuchElementException 抛出异常
        System.out.println(queue.remove());
    }

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 6]

*  不会抛出异常，有返回值
    
        /**
         * 不会抛出异常，有返回值
         */
        @Test
        public void test02(){
              
            //需要初始化队列的大小
            ArrayBlockingQueue queue = new ArrayBlockingQueue<>(3);
            System.out.println(queue.offer("cvzhanshi1"));
            System.out.println(queue.offer("cvzhanshi2"));
            System.out.println(queue.offer("cvzhanshi3"));
            //添加 一个不能添加的元素 使用offer只会返回false 不会抛出异常
            System.out.println(queue.offer("cvzhanshi4"));
            System.out.println(queue.poll());
            System.out.println(queue.poll());
            System.out.println(queue.poll());
            //弹出 如果没有元素 只会返回null 不会抛出异常
            System.out.println(queue.poll());
        }
    
    ![在这里插入图片描述][ce90394abf87458daa75e04bfc7b0cb8.png]
 *  阻塞 等待
    
        /**
             * 阻塞 等待
             */
        @Test
        public void test03() throws InterruptedException {
              
            //需要初始化队列的大小
            ArrayBlockingQueue queue = new ArrayBlockingQueue<>(3);
        
            //一直阻塞 不会返回
            queue.put("cvzhanshi1");
            queue.put("cvzhanshi2");
            queue.put("cvzhanshi3");
        
            //如果队列已经满了， 再进去一个元素  这种情况会一直等待这个队列 什么时候有了位置再进去，程序不会停止
            queue.put("a");
        
            System.out.println(queue.take());
            System.out.println(queue.take());
            System.out.println(queue.take());
            //如果我们再来一个  这种情况也会等待，程序会一直运行 阻塞
            System.out.println(queue.take());
        }
 *  超时 等待
    
        /**
             * 等待 超时阻塞
             *  这种情况也会等待队列有位置 或者有产品 但是会超时结束
             */
        @Test
        public void test4() throws InterruptedException {
              
            ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
            blockingQueue.offer("a");
            blockingQueue.offer("b");
            blockingQueue.offer("c");
            System.out.println("开始等待");
            blockingQueue.offer("d",2, TimeUnit.SECONDS);  //超时时间2s 等待如果超过2s就结束等待
            System.out.println("结束等待");
            System.out.println("===========取值==================");
            System.out.println(blockingQueue.poll());
            System.out.println(blockingQueue.poll());
            System.out.println(blockingQueue.poll());
            System.out.println("开始等待");
            blockingQueue.poll(2,TimeUnit.SECONDS); //超过两秒 我们就不要等待了
            System.out.println("结束等待");
        }
    
    ![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 7]

### 10.3 SynchronousQueue同步队列 ###

**没有容量，进去一个元素，必须等待取出来之后，才能再往里面放一个元素**

代码示例：

/**
     * 同步队列
     * 和其他的BlockingQueue不一样，SynchronousQueue不存储元素
     * put了一个元素，必须从里面先take取出来，否则不能在put进去值
     */
    public class SynchronousQueueDemo {
        
        public static void main(String[] args) {
        
            BlockingQueue<String> queue = new SynchronousQueue<>();  //同步队列
    
            //使用两个进程
            // 一个进程 放进去
            // 一个进程 拿出来
            new Thread(() -> {
        
                try {
        
                    System.out.println(Thread.currentThread().getName()+" Put 1");
                    queue.put("1");
                    System.out.println(Thread.currentThread().getName()+" Put 2");
                    queue.put("2");
                    System.out.println(Thread.currentThread().getName()+" Put 3");
                    queue.put("3");
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
    
    
            },"A").start();
            new Thread(() -> {
        
                try {
        
                    System.out.println(Thread.currentThread().getName()+" Take "+queue.take());
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println(Thread.currentThread().getName()+" Take "+queue.take());
                    TimeUnit.SECONDS.sleep(3);
                    System.out.println(Thread.currentThread().getName()+" Take "+queue.take());
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            },"B").start();
        }
    }

![在这里插入图片描述][f01547e789274029aa7a16b76e099716.png]

## 11、线程池 ##

线程池：**三大方法、七大参数、四种拒绝策略**

下面的示例业务图：

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 8]

> 池化技术

程序的运行的**本质就是占用系统的资源**，为了优化资源的使用 ==> **池化技术**（线程池、连接池、对象池、内存池）

**池化技术：事先准备好一些资源，有人要用，就到池里拿，用完之后还给池**

**线程池的好处**：

*  降低资源的消耗
 *  提高响应的速度
 *  方便管理

**线程复用、可以控制最大并发数、方便管理线程**

> **线程池：三大方法**

*  **ExecutorService threadPool = Executors.newSingleThreadExecutor(); //单个线程**
 *  **ExecutorService threadPool = Executors.newFixedThreadPool(5); //创建一个固定的线程池的大小**
 *  **ExecutorService threadPool = Executors.newCachedThreadPool(); //可伸缩的，遇强则强，遇弱则弱**

/**
     * @author cVzhanshi
     * @create 2021-09-22 20:23
     */
    public class Demo01 {
        
        public static void main(String[] args) {
        
    //        ExecutorService threadPool = Executors.newSingleThreadExecutor(); 	//单个线程
    //        ExecutorService threadPool = Executors.newFixedThreadPool(5);         //创建一个固定的线程池的大小
            ExecutorService threadPool = Executors.newCachedThreadPool();      //可伸缩的，遇强则强，遇弱则弱
            try {
        
                for (int i = 0; i < 10; i++) {
        
                    threadPool.execute(() -> {
        
                        System.out.println(Thread.currentThread().getName()+ " ok");
                    });
                }
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                threadPool.shutdown();
            }
        }
    }

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 9]

> **线程池：七大参数**

源码分析

public static ExecutorService newSingleThreadExecutor() {
        
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    
    public static ExecutorService newFixedThreadPool(int nThreads) {
        
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    
    public static ExecutorService newCachedThreadPool() {
        
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
    // 本质：调用ThreadPoolExecutor()
    // 分析源码
    public ThreadPoolExecutor(int corePoolSize,      // 核心线程池大小
                              int maximumPoolSize,   // 最大核心线程池大小
                              long keepAliveTime,	 // 超时了没有人用就会释放
                              TimeUnit unit,		 // 超时单位
                              BlockingQueue<Runnable> workQueue, // 阻塞队列
                              ThreadFactory threadFactory,		 // 线程工厂，创建线程的，一般不用动
                              RejectedExecutionHandler handler   // 拒绝策略) {
        
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 10]

**对于Integer.MAX\_VALUE初始值较大，所以一般情况我们要使用底层的ThreadPoolExecutor来创建线程池**

> 自定义线程池

/**
     * @author cVzhanshi
     * @create 2021-09-22 20:23
     */
    
    /**
     * - new ThreadPoolExecutor.AbortPolicy()： => 该拒绝策略为：银行满了，还有人进来，不处理这个人的，并抛出异常，超出最大承载，就会抛出异常：队列容量大小+maxPoolSize
     * - new ThreadPoolExecutor.CallerRunsPolicy()：=> 该拒绝策略为：哪来的去哪里 main线程进行处理
     * - new ThreadPoolExecutor.DiscardPolicy():   => 该拒绝策略为：队列满了,丢掉异常，不会抛出异常。
     * - new ThreadPoolExecutor.DiscardOldestPolicy()：=> 该拒绝策略为：队列满了，尝试去和最早的进程竞争，不会抛出异常
     */
    public class Demo02 {
        
        public static void main(String[] args) {
        
            ExecutorService threadPool = new ThreadPoolExecutor(
                    2,
                    5,
                    3,
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(3),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.AbortPolicy()
            );
            try {
        
                for (int i = 0; i < 10; i++) {
        
                    threadPool.execute(() -> {
        
                        System.out.println(Thread.currentThread().getName()+ " ok");
                    });
                }
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                threadPool.shutdown();
            }
        }
    }

> **线程池：四种拒绝策略**

![在这里插入图片描述][fdb3be97928d402ba7ef2f84be9fd0e1.png]

*  new ThreadPoolExecutor.AbortPolicy()： => 该拒绝策略为：银行满了，还有人进来，不处理这个人的，并抛出异常，超出最大承载，就会抛出异常：队列容量大小+maxPoolSize
 *  new ThreadPoolExecutor.CallerRunsPolicy()：=> 该拒绝策略为：哪来的去哪里 main线程进行处理
 *  new ThreadPoolExecutor.DiscardPolicy(): => 该拒绝策略为：队列满了,丢掉异常，不会抛出异常。
 *  new ThreadPoolExecutor.DiscardOldestPolicy()：=> 该拒绝策略为：队列满了，尝试去和最早的进程竞争，不会抛出异常

> 小结与拓展

如何去设置线程池的最大大小如何去设置？

CPU密集型和IO密集型！

*  **CPU密集型：电脑的核数是几核就选择几；选择maximunPoolSize的大小**
 *  \*\*I/O密集型：\*\*在程序中有15个大型任务，io十分占用资源；I/O密集型就是判断我们程序中十分耗I/O的线程数量，线程池大小大约是最大I/O数的一倍到两倍之间。

Runtime.getRuntime().availableProcessors()  // 获取CPU的核数

## 12、四大函数式接口 ##

> 函数式接口：有且只有一个方法的接口

// Runnable
    @FunctionalInterface
    public interface Runnable {
        
        public abstract void run();
    }
    //超级多的@FunctionalInterface
    //简化编程模型，在新版本的框架底层大量应用
    //foreach()的参数也是一个函数式接口，消费者类的函数式接口

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_17_color_FFFFFF_t_70_g_se_x_16]

函数型接口可以使用lambda表达式

**代码测试**：

> **Function函数式型接口：有一个输入参数，有一个输出参数**

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 11]

/**
     * @author cVzhanshi
     * @create 2021-09-23 9:35
     */
    /**
     * Function 函数型接口，有一个输入参数，有一个输出参数
     * 只要是 函数型接口都可以用lambda表达式简化
     */
    public class Demo01 {
        
        public static void main(String[] args) {
        
    //        Function<String, String> function = new Function<String, String>() {
        
    //            @Override
    //            public String apply(String str) {
        
    //                return str;
    //            }
    //        };
            Function<String, String> function = (str) -> {
        return str;};
            System.out.println(function.apply("cvzhanshi"));
        }
    }

> **Predicate断定型接口：只有一个输入参数，返回值只能是 布尔值**

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 12]

/**
     * @author cVzhanshi
     * @create 2021-09-23 9:42
     */
    
    /**
     * 断定型接口：只有一个输入参数，返回值只能是 布尔值
     */
    public class Demo02 {
        
        public static void main(String[] args) {
        
            // 判断字符串是否为空
    //        Predicate<String> predicate = new Predicate<String>(){
        
    //            @Override
    //            public boolean test(String str) {
        
    //                return str.isEmpty();
    //            }
    //        };
            Predicate<String> predicate = (str) -> {
        return str.isEmpty();};
            System.out.println(predicate.test("cvzhanshi"));
        }
    }

> **Consumer 消费型接口：只有输入，没有返回值**

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 13]

/**
     * Consumer 消费型接口：只有输入，没有返回值
     */
    public class Demo03 {
        
        public static void main(String[] args) {
        
    //        Consumer<String> consumer = new Consumer<String>(){
        
    //            @Override
    //            public void accept(String str) {
        
    //                System.out.println(str);
    //            }
    //        };
    
            Consumer<String> consumer = (str) -> {
         System.out.println(str);};
            consumer.accept("cvzhanshi");
        }
    }

> Supplier 供给型接口 没有参数，只有返回值

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_18_color_FFFFFF_t_70_g_se_x_16]

/**
     * @author cVzhanshi
     * @create 2021-09-23 9:52
     */
    
    /**
     * Supplier 供给型接口 没有参数，只有返回值
     */
    public class Demo04 {
        
        public static void main(String[] args) {
        
    //        Supplier<Integer> supplier = new Supplier<Integer>() {
        
    //            @Override
    //            public Integer get() {
        
    //                return 1024;
    //            }
    //        };
    
            Supplier<Integer> supplier = () -> {
         return 1024; };
            System.out.println(supplier.get());
        }
    }

## 13、Stream流式计算 ##

> 什么是Stream流式计算？

大数据：存储 + 计算

**集合、MySQL的本质就是存储**

计算都应该交给**流**来操作

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 14]

代码示例：

题目要求：一分钟内完成此题，只能用一行代码实现！

现在有6个用户！筛选

*  ID必须是偶数
 *  年龄必须大于23
 *  用户名转为大写
 *  用户名字母倒序排序
 *  只输出一个用户

/**
     * @author cVzhanshi
     * @create 2021-09-23 10:40
     */
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public class User {
        
        private Integer id;
        private String name;
        private Integer age;
    }
    
    /**
     * @author cVzhanshi
     * @create 2021-09-23 10:42
     */
    
    /**
     * 题目要求：一分钟内完成此题，只能用一行代码实现！
     * 现在有6个用户！筛选
     * 1、ID必须是偶数
     * 2、年龄必须大于23
     * 3、用户名转为大写
     * 4、用户名字母倒序排序
     * 5、只输出一个用户
     */
    public class Test {
        
        public static void main(String[] args) {
        
            User user1 = new User(1,"a",21);
            User user2 = new User(2,"b",22);
            User user3 = new User(3,"c",23);
            User user4 = new User(4,"d",24);
            User user5 = new User(5,"e",25);
            User user6 = new User(6,"f",26);
    
            // 集合就是存储
            List<User> list = Arrays.asList(user1, user2, user3, user4, user5, user6);
    
            // 计算交给Stream流
            list.stream()
                    .filter(user -> {
        return user.getId() % 2 == 0;})
                    .filter(user -> {
        return user.getAge() > 23;})
                    .map(user -> {
        return new User(user.getId(),user.getName().toUpperCase(),user.getAge());})
                    .sorted((u1,u2) -> {
        return u2.getName().compareTo(u1.getName());})
                    .limit(1)
                    .forEach(System.out::println);
        }
    }

![在这里插入图片描述][c2698f764d5343abb922808becbbd10d.png]

## 14、ForkJoin ##

> 什么是ForkJoin

ForkJoin**主要用于并发执行任务，提高效率，大数据量**。

**大任务分成小任务，最后把小结果合并成大结果**  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 15]

> ForkJoin 特点：工作窃取

**ForkJoin 维护的是双端队列**

**A、B线程执行任务，B更快执行完，他会去A那窃取任务过来执行，可以提高效率**

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_15_color_FFFFFF_t_70_g_se_x_16]

> ForkJoin 操作

查看ForkJoinPool

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 16]

查看ForkJoinTask  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 17]

查看直接子类RecursiveTask或RecursiveAction，可以看到使用方式

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 18]

如何使用forkjoin

*  ForkJoinPool 通过它来执行
 *  计算任务 ForkJoin .execute(ForkJoinTask task)
 *  计算类要继承ForkJoinTask

代码测试：

计算类

/**
     * @author cVzhanshi
     * @create 2021-09-23 11:00
     */
    /**
     * 求和计算的任务
     * 如何使用forkjoin
     * 1、forkjoinPool  通过他来执行
     * 2、计算任务 forkjoinPool.execute(ForkJoinTask task)
     * 3、计算类要继承ForkJoinTask
     */
    public class ForkJoinDemo extends RecursiveTask<Long> {
        
        private Long start;
        private Long end;
    
        private Long temp = 100000L;// 切割任务的临界值
    
        public ForkJoinDemo(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 {
        
                //使用forkJoin 分而治之 计算
                //计算平均值
                long middle = (start + end) / 2;
                ForkJoinDemo leftTask = new ForkJoinDemo(start, middle);
                leftTask.fork();  //拆分任务，把线程任务压入线程队列
                ForkJoinDemo rightTask = new ForkJoinDemo(middle, end);
                rightTask.fork();  //拆分任务，把线程任务压入线程队列
                long taskSum = leftTask.join() + rightTask.join();
                return taskSum;
            }
        }
    }

测试类

/**
     * @author cVzhanshi
     * @create 2021-09-23 11:57
     */
    public class Test {
        
    
        // 普通的计算
        @org.junit.Test
        public void test01(){
        
            Long sum = 0L;
            long start = System.currentTimeMillis();
            for (Long i = 1L; i <= 10_0000_0000; i++) {
        
                sum += i;
            }
            long end = System.currentTimeMillis();
            System.out.println("sum = " + sum + " 时间：" + (end - start)); // 8543ms
        }
    
        // ForkJoin的计算
        @org.junit.Test
        public void test02() throws ExecutionException, InterruptedException {
        
            long start = System.currentTimeMillis();
            ForkJoinDemo task = new ForkJoinDemo(0L, 10_0000_0000L);
            ForkJoinPool joinPool = new ForkJoinPool();
            ForkJoinTask<Long> submit = joinPool.submit(task);
            Long sum = submit.get();
            long end = System.currentTimeMillis();
            System.out.println("sum = " + sum + " 时间：" + (end - start)); // 5623ms
        }
    
    
        // Stream并行流计算
        @org.junit.Test
        public void test03() {
        
            long start = System.currentTimeMillis();
            long sum = LongStream.rangeClosed(0L, 10_0000_0000L).parallel().reduce(0, Long::sum);
            long end = System.currentTimeMillis();
            System.out.println("sum = " + sum + " 时间：" + (end - start)); // 910ms
        }
    }

## 15、异步回调 ##

> Future 设计的初衷：对将来的某个事件的结果进行建模

类似于是前端 --> 发送ajax异步请求给后端

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 19]

一般都使用**CompletableFuture**

> **没有返回值的runAsync异步回调**

/**
     * @author cVzhanshi
     * @create 2021-09-23 14:38
     */
    public class Demo01 {
        
        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(Thread.currentThread().getName() + "runAsync=>Void");
            });
            System.out.println("111111111111111111111111");
            completableFuture.get(); // 获取阻塞执行结果
        }
    }

![在这里插入图片描述][d4691824c6bc450797dae1e032f5bc4e.png]

> **有返回值的异步回调supplyAsync**

/**
     * @author cVzhanshi
     * @create 2021-09-23 14:38
     */
    public class Demo01 {
        
        public static void main(String[] args) throws ExecutionException, InterruptedException {
        
            // 发起一个请求
            CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
        
                System.out.println(Thread.currentThread().getName() + "supplyAsync=>Integer");
                int i = 10 / 0;
                return 1024;
            });
            // 得到返回值
            Integer result = completableFuture.whenComplete((t, u) -> {
        
                System.out.println("t => " + t);  // 正常的返回结果
                System.out.println("u => " + u);  // 错误信息：java.util.concurrent.CompletionException:java.lang.ArithmeticException: / by zero
            }).exceptionally((e) -> {
        
                System.out.println(e.getMessage());
                return 5000;
            }).get();
    
            System.out.println(result);
    
        }
    }

*  正常的返回结果
    
    ![在这里插入图片描述][031eba70979d4bc5b533157ffa2b0946.png]
 *  出现异常，返回错误信息
    
    ![在这里插入图片描述][734310ec7efc44d08eab4849a69db180.png]

**whenComplete**: 有两个参数，一个是t 一个是u

T：是代表的 **正常返回的结果**

U：是代表的 **抛出异常的错误信息**

如果发生了异常，get可以获取到**exceptionally**返回的值

> [并发编程JUC（下）][JUC]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/dc0b4fb9cf494abf99a8661f2bf35cbb.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/c217f5edb448418da7b66025fd9fabb2.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 2]: https://img-blog.csdnimg.cn/d4c16ddd02434429a54300da72bfd96c.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 3]: https://img-blog.csdnimg.cn/3d707fc848f042199b7036ff41b33324.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 4]: https://img-blog.csdnimg.cn/ccdbdd516bb34d6885a25b7c09dcb81c.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 5]: https://img-blog.csdnimg.cn/8dd8edba4fbc43c5a884915a25357159.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 6]: https://img-blog.csdnimg.cn/d8728611f648431aab344fa3fe6e90d8.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[ce90394abf87458daa75e04bfc7b0cb8.png]: https://img-blog.csdnimg.cn/ce90394abf87458daa75e04bfc7b0cb8.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 7]: https://img-blog.csdnimg.cn/de757818af7342f491d277cf93c62d36.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[f01547e789274029aa7a16b76e099716.png]: https://img-blog.csdnimg.cn/f01547e789274029aa7a16b76e099716.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 8]: https://img-blog.csdnimg.cn/fb20a4cc3e1a4d3c888a399822986a76.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 9]: https://img-blog.csdnimg.cn/9a18adfc8962434c8ebe159e674a0757.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 10]: https://img-blog.csdnimg.cn/1ea4a93fccfe410aa4795b33c2654690.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[fdb3be97928d402ba7ef2f84be9fd0e1.png]: https://img-blog.csdnimg.cn/fdb3be97928d402ba7ef2f84be9fd0e1.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_17_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/ffcc3d89244841dd9aeb5cee753e272b.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_17,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 11]: https://img-blog.csdnimg.cn/f9c6976f56814a8d8aefbd79aae9325c.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 12]: https://img-blog.csdnimg.cn/a234148c4f874eacbc39b7645cb0c454.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 13]: https://img-blog.csdnimg.cn/b62898423b4f437ea903fba051f87d8e.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_18_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/88218dc3d3f1450c9ce48b86f783e5ed.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_18,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 14]: https://img-blog.csdnimg.cn/ce2fd141fb9f4d0596525e1586919172.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[c2698f764d5343abb922808becbbd10d.png]: https://img-blog.csdnimg.cn/c2698f764d5343abb922808becbbd10d.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 15]: https://img-blog.csdnimg.cn/c705763368dd4991a6d2c75f70e9289c.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_15_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/ec776135c8814e059c671b03a5854514.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_15,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 16]: https://img-blog.csdnimg.cn/5a11903d4c1a47cbba7ba3fd9c8e1a7b.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 17]: https://img-blog.csdnimg.cn/dc094ab2d49e4321bedd970d1d291c13.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 18]: https://img-blog.csdnimg.cn/6742e0d881714bcb85cd0b67778fd8b5.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAY3blsZXnpLo_size_20_color_FFFFFF_t_70_g_se_x_16 19]: https://img-blog.csdnimg.cn/5f9478555e4e4bf38e6c0d31fb52569e.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAY3blsZXnpLo=,size_20,color_FFFFFF,t_70,g_se,x_16
[d4691824c6bc450797dae1e032f5bc4e.png]: https://img-blog.csdnimg.cn/d4691824c6bc450797dae1e032f5bc4e.png
[031eba70979d4bc5b533157ffa2b0946.png]: https://img-blog.csdnimg.cn/031eba70979d4bc5b533157ffa2b0946.png
[734310ec7efc44d08eab4849a69db180.png]: https://img-blog.csdnimg.cn/734310ec7efc44d08eab4849a69db180.png
[JUC]: https://blog.csdn.net/qq_45408390/article/details/120502412