Java中Juc并发编程基础

青旅半醒 2022-10-23 01:59 258阅读 0赞

1. 什么是JUC

> 就是java.util.concurrent并发包下面使用的工具包

## 1.1 线程和进程 ##

**进程:** 是一个程序,QQ.exe,网易云音乐 ,大数据领域的NameNode其实就是程序的集合,一个进程往往可以包含多个线程,至少包含一个线程!

那么我们的Java默认有几个线程呢? main 和 gc线程

**线程:**  例如我们的360我们可以进行杀毒期间也可以进行清理数据

**Java 真的可以开启线程吗 ?** 开不了,线程是本地方法来负责开启的

> 并行和并发

**并发:** 多线程操作同一个资源

*  CPU单核,模拟出来多线程,快速交替

**并行:** 多个人一起行走

*  CPU多核,多个线程可以同时运行

**并发编程的本质就是充分利用CPU的资源**

## 1.2 多线程 ##

如有需要请看我博客

https://blog.csdn.net/weixin\_43704599/article/details/107379994

# 2. Lock锁 #

## 2.1 synchronized ##

> synchronized

package com.bigdata.juc.demo01;
    
    /**
    * Copyright (c) 2020 bigdata ALL Rights Reserved
    * Project: learning
    * Package: com.bigdata.juc.demo01
    * Version: 1.0
    * 真正的多线程开发,公司的开发
    * 线程就是一个单独的资源类,没有任何附属的操作
    * 1. 属性,方法
    *
    * @author qingzhi.wu 2020/7/16 17:21
    */
    public class SaleTicketDemo01 {
       public static void main(String[] args) {
           //并发:多线程操作同一个资源类
           Ticket ticket = new Ticket();
           new Thread(() -> {
               for (int i = 0; i < 40; i++) {
    
                   ticket.sale();
    
    
              }
          }, "A").start();
    
           new Thread(() -> {
               for (int i = 0; i < 40; i++) {
    
                   ticket.sale();
    
    
              }
          }, "B").start();
    
           new Thread(() -> {
               for (int i = 0; i < 40; i++) {
    
                   ticket.sale();
    
    
              }
          }, "C").start();
      }
    }
    
    //资源类 OOP
    class Ticket {
    
       //属性,方法
       private int number = 30;
    
       //卖票的方式
       public synchronized void sale() {
           if (number > 0) {
               System.out.println(Thread.currentThread().getName() + "卖出了" + number-- + "票" + "剩余: " + number);
          }
      }
    }

## 2.2 RentrantLock ##

**公平锁: 十分公平,先来后到**

**非公平锁: 十分不公平,可以插队(默认)**

//并发:多线程操作同一个资源类
           Ticket ticket = new Ticket();
           new Thread(() -> {
               for (int i = 0; i < 40; i++) {
    
                   ticket.sale();
    
    
              }
          }, "A").start();
    
           new Thread(() -> {
               for (int i = 0; i < 40; i++) {
    
                   ticket.sale();
    
    
              }
          }, "B").start();
    
           new Thread(() -> {
               for (int i = 0; i < 40; i++) {
    
                   ticket.sale();
    
    
              }
          }, "C").start();

## 2.3 synchronized 和 lock 的区别 ##

1.  synchronized  是一个内置的Java关键字,Lock是一个java类
2.  synchronized  无法判断获取锁的状态,Lock 可以判断是否获取到了锁
3.  synchronized  会自动释放锁,Lock 必须要手动释放锁!如果不释放锁,**死锁**
4.  synchronized  线程1(获得锁,阻塞) 线程2 (等待,傻傻的等) Lock锁就不一定会傻傻的等了
5.  synchronized  可重入锁,不可以中断的,非公平;Lock ,可重入锁,可以判断锁,非公平(可以自己设置)
6.  synchronized  适合锁少量的代码同步问题,Lock 适合锁大量的同步代码!

# 3. 生产者和消费者问题 #

## 3.1 生产者和消费者问题 synchronized 版本 ##

package com.bigdata.juc.pc;
    
    /**
    * Copyright (c) 2019 bigdata ALL Rights Reserved
    * Project: learning
    * Package: com.bigdata.juc.pc
    * Version: 1.0
    * 线程之间的通信问题
    * 线程交替执行
    * 等待唤醒,通知唤醒
    * A B 线程 操作同一个变量 num = 0
    * A num +1
    * B num -1
    * @author qingzhi.wu 2020/7/16 20:55
    */
    public class Test1 {
       public static void main(String[] args) {
           Data data = new Data();
    
           new Thread(()->{
               for (int i = 0; i < 10; i++) {
                   data.increment();
              }
    
          },"A").start();
    
           new Thread(()->{
               for (int i = 0; i < 10; i++) {
                   data.decrement();
              }
    
          },"B").start();
      }
    }
    
    //判断等待,业务,通知
    //资源类
    class Data {
       private int number = 0;
       //+1
       public synchronized void  increment(){
           if (number != 0){
               try {
                   this.wait();
              } catch (InterruptedException e) {
                   e.printStackTrace();
              }
          }
           number++;
           System.out.println(Thread.currentThread().getName()+"=>"+number);
           //通知其他线程,我+1完毕了
           this.notifyAll();
      }
       //-1
       public synchronized void  decrement(){
           if (number == 0){
               //等待
               try {
                   this.wait();
              } catch (InterruptedException e) {
                   e.printStackTrace();
              }
          }
           number--;
           System.out.println(Thread.currentThread().getName()+"=>"+number);
           // 通知其他线程 ,我 -1 完毕了
           this.notifyAll();
      }
    }

> 问题存在,A B C D 4个线程怎么办,改为while循环就好了-- 虚假唤醒问题
> 
> https://www.cnblogs.com/tqyysm/articles/9765667.html

package com.bigdata.juc.pc;
    
    /**
    * Copyright (c) 2019 bigdata ALL Rights Reserved
    * Project: learning
    * Package: com.bigdata.juc.pc
    * Version: 1.0
    * 线程之间的通信问题
    * 线程交替执行
    * 等待唤醒,通知唤醒
    * A B 线程 操作同一个变量 num = 0
    * A num +1
    * B num -1
    * @author qingzhi.wu 2020/7/16 20:55
    */
    public class Test1 {
       public static void main(String[] args) {
           Data data = new Data();
    
           new Thread(()->{
               for (int i = 0; i < 10; i++) {
                   data.increment();
              }
    
          },"A").start();
    
           new Thread(()->{
               for (int i = 0; i < 10; i++) {
                   data.decrement();
              }
    
          },"B").start();
    
           new Thread(()->{
               for (int i = 0; i < 10; i++) {
                   data.increment();
              }
    
          },"C").start();
           new Thread(()->{
               for (int i = 0; i < 10; i++) {
                   data.decrement();
              }
    
          },"D").start();
      }
    }
    
    //判断等待,业务,通知
    //资源类
    class Data {
       private int number = 0;
       //+1
       public synchronized void  increment(){
           while (number != 0){
               try {
                   this.wait();
              } catch (InterruptedException e) {
                   e.printStackTrace();
              }
          }
           number++;
           System.out.println(Thread.currentThread().getName()+"=>"+number);
           //通知其他线程,我+1完毕了
           this.notifyAll();
      }
       //-1
       public synchronized void  decrement(){
           while (number == 0){
               //等待
               try {
                   this.wait();
              } catch (InterruptedException e) {
                   e.printStackTrace();
              }
          }
           number--;
           System.out.println(Thread.currentThread().getName()+"=>"+number);
           // 通知其他线程 ,我 -1 完毕了
           this.notifyAll();
      }
    }

## 3.2 生产者和消费者问题 Lock版本 ##

package com.bigdata.juc.pc;
    
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
    * Copyright (c) 2019 bigdata ALL Rights Reserved
    * Project: learning
    * Package: com.bigdata.juc.pc
    * Version: 1.0
    * 线程之间的通信问题
    * 线程交替执行
    * 等待唤醒,通知唤醒
    * A B 线程 操作同一个变量 num = 0
    * A num +1
    * B num -1
    *
    * @author qingzhi.wu 2020/7/16 20:55
    */
    public class Test2 {
       public static void main(String[] args) {
           Data data = new Data();
    
           new Thread(() -> {
               for (int i = 0; i < 10; i++) {
                   data.increment();
              }
    
          }, "A").start();
    
           new Thread(() -> {
               for (int i = 0; i < 10; i++) {
                   data.decrement();
              }
    
          }, "B").start();
    
           new Thread(() -> {
               for (int i = 0; i < 10; i++) {
                   data.increment();
              }
    
          }, "C").start();
           new Thread(() -> {
               for (int i = 0; i < 10; i++) {
                   data.decrement();
              }
    
          }, "D").start();
      }
    }
    
    //判断等待,业务,通知
    //资源类
    class Data2 {
       Lock lock = new ReentrantLock();
       Condition condition = lock.newCondition();
    
       private int number = 0;
    
       //+1
       public void increment() {
           lock.lock();
           try {
               while (number != 0) {
                   try {
                       condition.await();
                  } catch (InterruptedException e) {
                       e.printStackTrace();
                  }
              }
               number++;
               System.out.println(Thread.currentThread().getName() + "=>" + number);
               //通知其他线程,我+1完毕了
               condition.signalAll();
          } finally {
               lock.unlock();
          }
    
      }
    
       //-1
       public void decrement() {
    
           lock.lock();
           try {
               while (number == 0) {
                   //等待
                   try {
                       condition.await();
                  } catch (InterruptedException e) {
                       e.printStackTrace();
                  }
              }
               number--;
               System.out.println(Thread.currentThread().getName() + "=>" + number);
               // 通知其他线程 ,我 -1 完毕了
               condition.signalAll();
          } finally {
               lock.unlock();
          }
      }
    }

## 3.3 生产者和消费者问题(精确通知) ##

package com.bigdata.juc.pc;
    
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
    * Copyright (c) 2019 bigdata ALL Rights Reserved
    * Project: learning
    * Package: com.bigdata.juc.pc
    * Version: 1.0
    * 线程之间的通信问题
    * 线程交替执行
    * 等待唤醒,通知唤醒
    * A B 线程 操作同一个变量 num = 0
    * A num +1
    * B num -1
    * A-B-C-A-B
    * @author qingzhi.wu 2020/7/16 20:55
    */
    public class Test3 {
       public static void main(String[] args) {
           Data3 data = new Data3();
    
           new Thread(() -> {
               for (; ; ) {
                   data.printA();
              }
    
          }, "A").start();
    
           new Thread(() -> {
               for (; ; ) {
                   data.printB();
              }
    
          }, "B").start();
    
           new Thread(() -> {
               for (; ; ) {
                   data.printC();
              }
    
          }, "C").start();
      }
    }
    
    //判断等待,业务,通知
    //资源类
    class Data3 {
       private Lock lock = new ReentrantLock();
       private Condition condition1 = lock.newCondition();
       private Condition condition2 = lock.newCondition();
       private Condition condition3 = lock.newCondition();
       private int number = 1; //2 3
       public void printA(){
           lock.lock();
           try {
               while (number != 1){
                   //等待
                   try {
                       condition1.await();
                  } catch (InterruptedException e) {
                       e.printStackTrace();
                  }
              }
               System.out.println(Thread.currentThread().getName());
               number = 2;
               condition2.signal();
          }finally {
               lock.unlock();
          }
      }
       public void printB(){
           lock.lock();
           try {
               while (number != 2){
                   try {
                       condition2.await();
                  } catch (InterruptedException e) {
                       e.printStackTrace();
                  }
              }
               number ++;
               System.out.println(Thread.currentThread().getName());
               condition3.signal();
          }finally {
               lock.unlock();
          }
      }
       public void printC(){
           lock.lock();
           try {
               while (number != 3){
                   try {
                       condition3.await();
                  } catch (InterruptedException e) {
                       e.printStackTrace();
                  }
              }
               number = 1 ;
               System.out.println(Thread.currentThread().getName());
               condition1.signal();
               condition3.signal();
          }finally {
               lock.unlock();
          }
      }
    }

# 4. 八锁现象 #

> 如何判断锁的是谁?

## 4.1 单对象调用多方法 ##

package com.bigdata.lock8;
    
    import java.util.concurrent.TimeUnit;
    
    /**
    * Copyright (c) 2020 bigdata ALL Rights Reserved
    * Project: learning
    * Package: com.bigdata.lock8
    * Version: 1.0
    * 1. 标准情况下,两个线程先打印 发短信 还是打电话
    * 1 发短信
    * 2 打电话
    * @author qingzhi.wu 2020/7/17 9:30
    */
    public class Test1 {
       public static void main(String[] args) throws InterruptedException {
           //其实 锁的 phone 谁先拿到 就是谁先执行
           Phone phone = new Phone();
           //是因为锁的原因
           new Thread(()->{
               phone.senSms();
          }).start();
           TimeUnit.SECONDS.sleep(1);
           new Thread(()->{
               phone.call();
          }).start();
    
      }
    
       static class Phone {
           //synchronized 锁的对象是方法的调用者
           //两个方法用的是同一个锁,谁先拿到谁执行
           public synchronized  void senSms(){
               try {
                   TimeUnit.SECONDS.sleep(1);
              } catch (InterruptedException e) {
                   e.printStackTrace();
              }
               System.out.println("发短信");
          }
    
           public synchronized  void call(){
               System.out.println("call");
          }
      }
    }

**锁的是一个对象,两个对象的锁是一样的,所以发短信先执行然后再打电话**

## 4.2 多对象对象调用多方法 ##

package com.bigdata.lock8;
    
    import java.util.concurrent.TimeUnit;
    
    /**
    * Copyright (c) 2020 bigdata ALL Rights Reserved
    * Project: learning
    * Package: com.bigdata.lock8
    * Version: 1.0
    * 1. 标准情况下,两个线程先打印 发短信 还是打电话
    * 1 发短信
    * 2 打电话
    * @author qingzhi.wu 2020/7/17 9:30
    */
    public class Test1 {
       public static void main(String[] args) throws InterruptedException {
           //其实 锁的 phone 谁先拿到 就是谁先执行
           Phone phone = new Phone();
           Phone phone1 = new Phone();
           //是因为锁的原因
           new Thread(()->{
               phone.senSms();
          }).start();
           TimeUnit.SECONDS.sleep(1);
           new Thread(()->{
               phone1.call();
          }).start();
    
      }
    
       static class Phone {
           //synchronized 锁的对象是方法的调用者
           //两个方法用的是同一个锁,谁先拿到谁执行
           public synchronized  void senSms(){
               try {
                   TimeUnit.SECONDS.sleep(1);
              } catch (InterruptedException e) {
                   e.printStackTrace();
              }
               System.out.println("sendsms");
          }
    
           public synchronized  void call(){
               System.out.println("call");
          }
      }
    }

**因为是new 了两个资源类,所以 锁不是同一个,所以 先 打电话再发短信**

## 4.3 单对象同时调用同步方法和非同步方法 ##

package com.bigdata.lock8;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.lock8
     * Version: 1.0
     * 1. 标准情况下,两个线程先打印 发短信 还是打电话
     * 1 发短信
     * 2 打电话
     * @author qingzhi.wu 2020/7/17 9:30
     */
    public class Test1 {
        public static void main(String[] args) throws InterruptedException {
            //其实 锁的 phone  谁先拿到 就是谁先执行
            Phone phone = new Phone();
           // Phone phone1 = new Phone();
            //是因为锁的原因
            new Thread(()->{
                phone.senSms();
            }).start();
            TimeUnit.SECONDS.sleep(1);
            new Thread(()->{
                phone.hello();
            }).start();
    
        }
    
        static class Phone {
            //synchronized 锁的对象是方法的调用者
            //两个方法用的是同一个锁,谁先拿到谁执行
            public synchronized  void senSms(){
                try {
                    TimeUnit.SECONDS.sleep(4);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("sendsms");
            }
    
            public synchronized  void call(){
                System.out.println("call");
            }
    
            public void hello(){
                System.out.println("hello!");
            }
        }
    }

**肯定是先hello 在发短信,因为 他根本就没锁啊 不需要等待,直接就可以执行**

## 4.4 多对象调用同步方法和非同步方法 ##

package com.bigdata.lock8;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.lock8
     * Version: 1.0
     * 1. 标准情况下,两个线程先打印 发短信 还是打电话
     * 1 发短信
     * 2 打电话
     * @author qingzhi.wu 2020/7/17 9:30
     */
    public class Test1 {
        public static void main(String[] args) throws InterruptedException {
            //其实 锁的 phone  谁先拿到 就是谁先执行
            Phone phone = new Phone();
            Phone phone1 = new Phone();
            //是因为锁的原因
            new Thread(()->{
                phone.senSms();
            }).start();
            TimeUnit.SECONDS.sleep(1);
            new Thread(()->{
                phone1.hello();
            }).start();
    
        }
    
        static class Phone {
            //synchronized 锁的对象是方法的调用者
            //两个方法用的是同一个锁,谁先拿到谁执行
            public synchronized  void senSms(){
                try {
                    TimeUnit.SECONDS.sleep(4);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("sendsms");
            }
    
            public synchronized  void call(){
                System.out.println("call");
            }
    
            public void hello(){
                System.out.println("hello!");
            }
        }
    }

**肯定是hello先执行,因为他没有锁,他俩根本就没有关系,发短信还睡了4秒**

## 4.5 单对象调用同步和静态同步方法 ##

package com.bigdata.lock8;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.lock8
     * Version: 1.0
     * 1. 标准情况下,两个线程先打印 发短信 还是打电话
     * 1 发短信
     * 2 打电话
     * @author qingzhi.wu 2020/7/17 9:30
     */
    public class Test1 {
        public static void main(String[] args) throws InterruptedException {
            //其实 锁的 phone  谁先拿到 就是谁先执行
            Phone phone = new Phone();
            Phone phone1 = new Phone();
            //是因为锁的原因
            new Thread(()->{
                phone.senSms();
            }).start();
            TimeUnit.SECONDS.sleep(1);
            new Thread(()->{
                phone1.call();
            }).start();
    
        }
    
        static class Phone {
            //synchronized 锁的对象是方法的调用者
            //两个方法用的是同一个锁,谁先拿到谁执行
            public synchronized  void senSms(){
                try {
                    TimeUnit.SECONDS.sleep(4);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("sendsms");
            }
    
            public static synchronized  void call(){
                System.out.println("call");
            }
    
            public void hello(){
                System.out.println("hello!");
            }
        }
    }

**先执行的打电话,后执行的发短信,因为静态修饰的同步方法,锁是类的Class模板,同步非静态方法的锁是对象,所以他俩的锁不是一个锁**

## 4.6 单对象调用双静态方法 ##

package com.bigdata.lock8;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.lock8
     * Version: 1.0
     * 1. 标准情况下,两个线程先打印 发短信 还是打电话
     * 1 发短信
     * 2 打电话
     * @author qingzhi.wu 2020/7/17 9:30
     */
    public class Test1 {
        public static void main(String[] args) throws InterruptedException {
            //其实 锁的 phone  谁先拿到 就是谁先执行
            Phone phone = new Phone();
            Phone phone1 = new Phone();
            //是因为锁的原因
            new Thread(()->{
                phone.senSms();
            }).start();
            TimeUnit.SECONDS.sleep(1);
            new Thread(()->{
                phone1.call();
            }).start();
    
        }
    
        static class Phone {
            //synchronized 锁的对象是方法的调用者
            //两个方法用的是同一个锁,谁先拿到谁执行
            public static synchronized  void senSms(){
                try {
                    TimeUnit.SECONDS.sleep(4);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("sendsms");
            }
    
            public static synchronized  void call(){
                System.out.println("call");
            }
    
            public void hello(){
                System.out.println("hello!");
            }
        }
    }

**肯定是先发短信再打电话,因为他俩的锁的对象是同一个,都是class文件**

## 4.7 双对象调用双静态方法 ##

package com.bigdata.lock8;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.lock8
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 9:49
     */
    public class Test4 {
    
        public static void main(String[] args) throws InterruptedException {
            //其实 锁的 phone  谁先拿到 就是谁先执行
            //两个对象 两个锁
            Phone phone = new Phone();
            Phone phone1 = new Phone();
            //是因为锁的原因
            new Thread(()->{
                phone.senSms();
            }).start();
            TimeUnit.SECONDS.sleep(1);
            new Thread(()->{
                phone1.call();
            }).start();
    
        }
    
        static class Phone {
            //synchronized 锁的对象是方法的调用者
            //两个方法用的是同一个锁,谁先拿到谁执行
            // static 锁的是字节码Class 模板
            public static synchronized  void senSms(){
                try {
                    TimeUnit.SECONDS.sleep(4);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("sendsms");
            }
    
            public static synchronized  void call(){
                System.out.println("call");
            }
            // 这里没有锁 不是同步方法,不受锁的影响
            public static void hello(){
                System.out.println("hello");
            }
        }
    }

**先发短信,后打电话,因为锁是同一个class字节码文件**

# 5. 集合类 #

## 5.1 CopyOnWriteArrayList ##

package com.bigdata.list;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.UUID;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.list
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 10:21
     */
    public class ListTest {
        public static void main(String[] args) {
            List<String> list = new ArrayList<>();
            
            for (int i = 0; i < 10000; i++) {
                new Thread(()->{
                    list.add(UUID.randomUUID().toString().substring(0,5));
                    System.out.println(list);
                },String.valueOf(i)).start();
            }
    
        }
    }

![图片][93271ca937744b8aa9b48d491e604f6a.png]

> 可以采用Vector解决

package com.bigdata.list;
    
    import java.util.List;
    import java.util.UUID;
    import java.util.Vector;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.list
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 10:21
     */
    public class ListTest {
        public static void main(String[] args) {
            List<String> list = new Vector<>();
    
            for (int i = 0; i < 10000; i++) {
                new Thread(()->{
                    list.add(UUID.randomUUID().toString().substring(0,5));
                    System.out.println(list);
                },String.valueOf(i)).start();
            }
    
        }
    }

> 采用Collections.synchronizedList解决

package com.bigdata.list;
    
    import java.util.*;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.list
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 10:21
     */
    public class ListTest {
        public static void main(String[] args) {
            List<String> list = Collections.synchronizedList(new ArrayList<String>());
    
            for (int i = 0; i < 10000; i++) {
                new Thread(()->{
                    list.add(UUID.randomUUID().toString().substring(0,5));
                    System.out.println(list);
                },String.valueOf(i)).start();
            }
    
        }
    }

> CopyOnWriteArrayList 解决

package com.bigdata.list;
    
    import java.util.*;
    import java.util.concurrent.CopyOnWriteArrayList;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.list
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 10:21
     */
    public class ListTest {
        public static void main(String[] args) {
            CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
    
            for (int i = 0; i < 10000; i++) {
                new Thread(()->{
                    list.add(UUID.randomUUID().toString().substring(0,5));
                    System.out.println(list);
                },String.valueOf(i)).start();
            }
    
        }
    }

> **CopyOnWrite** 写入时复制 COW 计算机程序设计领域的一种优化策略,多个线程调用的时候,在写入的时候避免覆盖,造成数据问题 读写分离

**CopyOnWriteArrayList 比 Vector 效率高很多**

![20210318102130147.gif][]

## 5.2 CopyOnWriteArraySet ##

> HashSet的实现多线程插入

package com.bigdata.list;
    
    import java.util.HashSet;
    import java.util.Set;
    import java.util.UUID;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.list
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 11:14
     */
    public class SetTest {
        public static void main(String[] args) {
            Set<String> list = new HashSet<>();
    
            for (int i = 0; i < 10000; i++) {
                new Thread(()->{
                    list.add(UUID.randomUUID().toString().substring(0,5));
                    System.out.println(list);
                },String.valueOf(i)).start();
            }
        }
    }

![20210318102130185.gif][]

> 基于Collections的

package com.bigdata.list;
    
    import java.util.Collections;
    import java.util.HashSet;
    import java.util.Set;
    import java.util.UUID;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.list
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 11:14
     */
    public class SetTest {
        public static void main(String[] args) {
            Set<String> list = Collections.synchronizedSet(new HashSet<String>());
    
            for (int i = 0; i < 10000; i++) {
                new Thread(()->{
                    list.add(UUID.randomUUID().toString().substring(0,5));
                    System.out.println(list);
                },String.valueOf(i)).start();
            }
        }
    }

> CopyOnWriteArraySet

package com.bigdata.list;
    
    import java.util.UUID;
    import java.util.concurrent.CopyOnWriteArraySet;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.list
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 11:14
     */
    public class SetTest {
        public static void main(String[] args) {
            CopyOnWriteArraySet<String> list = new CopyOnWriteArraySet();
    
            for (int i = 0; i < 10000; i++) {
                new Thread(() -> {
                    list.add(UUID.randomUUID().toString().substring(0, 5));
                    System.out.println(list);
                }, String.valueOf(i)).start();
            }
        }
    }

## 5.3 ConcurrentHashMap ##

> 常用的HashMap

package com.bigdata.list;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.UUID;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.list
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 11:48
     */
    public class MapTest {
        public static void main(String[] args) {
            Map<String,String> map = new HashMap<String,String>();
    
            for (int i = 0; i < 10000; i++) {
                new Thread(()->{
                    map.put(UUID.randomUUID().toString().substring(0,5),UUID.randomUUID().toString());
                    System.out.println(map);
                },String.valueOf(i)).start();
            }
    
        }
    }

![20210318102130183.gif][]

> ConcurrentHashMap 解决

package com.bigdata.list;
    
    import java.util.UUID;
    import java.util.concurrent.ConcurrentHashMap;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.list
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 11:48
     */
    public class MapTest {
        public static void main(String[] args) {
            ConcurrentHashMap<String,String> map = new ConcurrentHashMap<String,String>();
    
            for (int i = 0; i < 10000; i++) {
                new Thread(()->{
                    map.put(UUID.randomUUID().toString().substring(0,5),UUID.randomUUID().toString());
                    System.out.println(map);
                },String.valueOf(i)).start();
            }
    
        }
    }

# 6. Callable #

1.  可以有返回值
2.  可以抛出异常
3.  方法不同,run(),call()

> 代码测试一下

package com.bigdata.callable;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.FutureTask;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.callable
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 13:12
     */
    public class CallableTest {
    
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            FutureTask<String> futureTask = new FutureTask<>(new MyThread());
    
            new Thread(futureTask).start();
    
            String s = futureTask.get();
            System.out.println(s);
        }
    }
    
    class MyThread implements Callable<String> {
    
        @Override
        public String call() throws Exception {
            System.out.println("call()");
            return "123456";
        }
    }

细节:

1.  有缓存
2.  结果可能需要等待,会阻塞

# 7. 辅助工具类 #

## 7.1 CountDownLatch ##

package com.bigdata.add;
    
    import java.util.concurrent.CountDownLatch;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.add
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 14:45
     */
    public class CountDownLatchDemo {
        public static void main(String[] args) throws InterruptedException {
            //总数是6
            CountDownLatch countDownLatch = new CountDownLatch(11);
            countDownLatch.countDown(); //-1
    
            for (int i = 1; i <= 10; i++) {
                new Thread(()->{
                    System.out.println(Thread.currentThread().getName() + "Go Out");
                    countDownLatch.countDown(); //数量 -1
                },String.valueOf(i)).start();
            }
    
            countDownLatch.await();
    
           // countDownLatch.countDown();
            System.out.println("Close Door");
        }
    }

*  等所有线程执行完毕之后 就会执行Close Door

## 7.2 CyclicBarrier ##

package com.bigdata.add;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.add
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 15:38
     */
    public class CycBarrierTest {
        public static void main(String[] args) {
            CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{
                System.out.println("集齐了,七颗龙珠");
            });
    
            for (int i = 1; i <= 7; i++) {
                int finalI = i;
                new Thread(()->{
                    System.out.println(Thread.currentThread().getName()+"拿到了第"+ finalI + "颗龙珠");
                    try {
                        cyclicBarrier.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }).start();
            }
        }
    }

**主要用来判断执行了几步了.都执行完成 执行最终线程**

## 7.3 Semaphore (信号量) ##

package com.bigdata.add;
    
    import java.util.concurrent.Semaphore;
    import java.util.concurrent.TimeUnit;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.add
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 16:02
     */
    public class SemaphoreTest {
        public static void main(String[] args) {
            Semaphore semaphore = new Semaphore(3);
            for (int i = 1; i <=6; i++) {
                int finalI = i;
                new Thread(
                        ()->{
                            try {
                                semaphore.acquire();
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
    
                            System.out.println(Thread.currentThread().getName()+"->"+ finalI+"进入了车位");
                            try {
                                TimeUnit.SECONDS.sleep(2);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }finally {
                                semaphore.release();
                            }
                            System.out.println(Thread.currentThread().getName()+"->"+ finalI+"离开了车位");
    
                        }
                ).start();
            }
        }
    }

**用于限流**

# 8. 读写锁 #

package com.bigdata.add;
    
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.locks.ReentrantReadWriteLock;
    
    /**
     * Copyright (c) 2020 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.add
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/17 17:21
     */
    public class ReadWriteLock {
        public static void main(String[] args) {
            MyCache myCache = new MyCache();
    
            for (int i = 1; i <= 10; i++) {
                int finalI = i;
                new Thread(()->{
                    myCache.put(finalI,new Object());
    
                },String.valueOf(finalI)).start();
            }
    
            for (int i = 1; i <= 10; i++) {
                int finalI = i;
                new Thread(()->{
                    myCache.get(finalI);
    
                },String.valueOf(finalI)).start();
            }
    
    
        }
    }
    class MyCache {
        private Map<Integer,Object> data = new HashMap<>();
        private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    
        public void put(Integer key,Object value){
            lock.writeLock().lock();
            try {
                System.out.println(Thread.currentThread().getName()+"开始添加数据");
                data.put(key,value);
                System.out.println(Thread.currentThread().getName()+"添加数据OK");
    
            } finally {
                lock.writeLock().unlock();
            }
        }
    
    
        public void get(Integer key){
            lock.readLock().lock();
            try {
                System.out.println(Thread.currentThread().getName()+"开始查询数据");
                data.get(key);
                System.out.println(Thread.currentThread().getName()+"查询数据OK");
    
            } finally {
                lock.readLock().unlock();
            }
        }
    }

*  读--写 互斥
 *  读--读 共享
 *  写--写 互斥

**写锁是独占锁,读锁是共享锁**

# 9. 阻塞队列 #

## 9.1 BlockingQueue ##

![图片][cabb912869bd7d59aa2938acffde0852.png]

## 9.2 四组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</td> 
  </tr> 
  <tr> 
   <td>移除</td> 
   <td>remove</td> 
   <td>poll</td> 
   <td>take</td> 
   <td>poll</td> 
  </tr> 
  <tr> 
   <td>判断队列首</td> 
   <td>element</td> 
   <td>peek</td> 
   <td>-</td> 
   <td>-</td> 
  </tr> 
 </tbody> 
</table>

> 抛出异常

/**
         * 抛出异常
         */
        public void test1(){
            ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<String>(3);
            System.out.println(blockingQueue.add("a"));
            System.out.println(blockingQueue.add("b"));
            System.out.println(blockingQueue.add("c"));
           //System.out.println(blockingQueue.add("d"));
            //String remove = blockingQueue.remove();
            System.out.println(blockingQueue.remove());
            System.out.println(blockingQueue.element());
    
            System.out.println(blockingQueue.remove());
            System.out.println(blockingQueue.remove());
            //System.out.println(blockingQueue.remove());
        }

> 不抛出异常

/**
         * 不抛出异常
         */
        public void test2(){
            ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<String>(3);
            System.out.println(blockingQueue.offer("a"));
            System.out.println(blockingQueue.offer("b"));
            System.out.println(blockingQueue.offer("c"));
            System.out.println(blockingQueue.offer("d")); //false
            System.out.println(blockingQueue.element());
            System.out.println(blockingQueue.poll());
            System.out.println(blockingQueue.poll());
            System.out.println(blockingQueue.poll());
            System.out.println(blockingQueue.poll()); //null
        }

> 阻塞等待

/**
         * 阻塞等待
         */
        public void test3() throws InterruptedException {
            ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<String>(3);
            blockingQueue.put("a");
            blockingQueue.put("b");
            blockingQueue.put("c");
            blockingQueue.take();
            blockingQueue.take();
            blockingQueue.take();
            blockingQueue.take();
            blockingQueue.put("d");
        }

> 超时等待

/**
         * 阻塞等待
         */
        public void test4() throws InterruptedException {
            ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<String>(3);
            blockingQueue.offer("a");
            blockingQueue.offer("b");
            blockingQueue.offer("c");
            // 等待超过2秒就退出
            blockingQueue.offer("d", 2,TimeUnit.SECONDS);
            blockingQueue.poll(2,TimeUnit.SECONDS);
            blockingQueue.poll(2,TimeUnit.SECONDS);
            blockingQueue.poll(2,TimeUnit.SECONDS);
            //超过2秒就退出
            blockingQueue.poll(2,TimeUnit.SECONDS); 
    //        blockingQueue.take();
    //        blockingQueue.take();
    //        blockingQueue.take();
    //        blockingQueue.take();
    //        blockingQueue.put("d");
        }

## 9.3 SynchronousQueue ##

package com.bigdata.bq;
    
    import java.util.concurrent.SynchronousQueue;
    import java.util.concurrent.TimeUnit;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.bq
     * Version: 1.0
     * 同步队列
     * 和其他的blocking queue 不一样SynchronousQueue 不存储元素
     * put了一个元素,必须从里面先take出来 否则不能再put进去值!
     * @author qingzhi.wu 2020/7/18 10:58
     */
    public class Test2 {
        public static void main(String[] args) {
            SynchronousQueue<Object> synchronousQueue = new SynchronousQueue<>();
    
            new Thread(
                    ()->{
                        try {
                            System.out.println(Thread.currentThread().getName() +"put 1");
                            synchronousQueue.put("1");
                            System.out.println(Thread.currentThread().getName() +"put 2");
                            synchronousQueue.put("2");
                            System.out.println(Thread.currentThread().getName() +"put 3");
                            synchronousQueue.put("3");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
    
            ,"T1").start();
    
            new Thread(
                    ()->{
                        try {
                            TimeUnit.SECONDS.sleep(3);
                            System.out.println(Thread.currentThread().getName() +"get-->" + synchronousQueue.take());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
    
                    ,"T2").start();
        }
    
    }

# 10. 线程池 #

## 10.1 线程池简介 ##

线程池: 三大方法,7大参数,4中拒绝策略

> 池化技术

程序的运行,本质: 占用系统的资源!

**线程池,连接池,内存池,对象池..**

池化技术: 事先准备好一些资源,有人要用,就来我这里拿,用完之后还给我

**线程池的好处:**

1.  降低资源的消耗
2.  提高响应的速度
3.  方便管理

线程复用,可以控制最大的并发数,管理线程

![20210318102130152.gif][]

## 10.2 三大方法 ##

package com.bigdata.pool;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.pool
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 11:27
     *
     *
     * Exectors 工具类  3大方法
     * 使用了线程池之后,使用线程池来创建线程
     */
    public class Demo01 {
        public static void main(String[] args) {
            // 单个线程
            //ExecutorService executorService = Executors.newSingleThreadExecutor();
    
    //        // 创建一个固定的线程池大小
            ExecutorService threadpool = Executors.newFixedThreadPool(1000);
    //        //可伸缩的,遇强则强,遇弱则弱
    //        Executors.newCachedThreadPool();
    
            try {
                for (int i = 0; i < 1000; i++) {
    
                    threadpool.execute(()->{
                        System.out.println(Thread.currentThread().getName()+"ok");
                    });
                }
            } finally {
                threadpool.shutdown();
            }
    
        }
    }

## 10.3 七大参数 ##

> 源码分析

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.acc = System.getSecurityManager() == null ?
                    null :
                    AccessController.getContext();
            this.corePoolSize = corePoolSize;
            this.maximumPoolSize = maximumPoolSize;
            this.workQueue = workQueue;
            this.keepAliveTime = unit.toNanos(keepAliveTime);
            this.threadFactory = threadFactory;
            this.handler = handler;
        }

*  int corePoolSize                                              核心线程数大小
 *  int maximumPoolSize                                   最大线程数大小
 *  long keepAliveTime                                       超时了,没人调用就会被释放
 *  TimeUnit unit                                                 超时单位
 *  BlockingQueue<Runnable> workQueue 阻塞队列
 *  ThreadFactory threadFactory                     线程工厂,创建线程的,一般不用动
 *  RejectedExecutionHandler handler          拒绝策略

![20210318102130188.gif][]

![20210318102130183.gif][]

> 手动创建线程池

## 10.4 四种拒绝策略 ##

![20210318102130149.gif][]

*  AbortPolicy                  超出承载就会报错
 *  CallerRunsPolicy         哪里来的去哪里
 *  DiscardPolicy               队列满了 就会丢掉 不会抛出异常
 *  DiscardOldestPolicy    队列满了,尝试和最早的竞争,竞争失败 也不会抛出异常

## 10.5 IO密集型和CPU密集型 ##

![20210318102130184.gif][]

package com.bigdata.pool;
    
    import java.util.concurrent.Executors;
    import java.util.concurrent.LinkedBlockingDeque;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.pool
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 11:51
     * AbortPolicy      超出承载就会报错
     * CallerRunsPolicy 哪里来的去哪里
     * DiscardPolicy    队列满了 就会丢掉 不会抛出异常
     * DiscardOldestPolicy 队列满了,尝试和最早的竞争,竞争失败 也不会抛出异常
     */
    public class Demo02 {
        public static void main(String[] args) {
            // 银行满了,还有人进来,不处理这个人的了,抛出异常
            /**
             * 最大线程到底该如何定义
             * 1. CPU 密集型 Runtime.getRuntime().availableProcessors()
             * 2. IO  密集型 判断你程序中十分耗费IO的线程 一般会设置 两倍
             *        程序  15个大型任务 IO 十分占用资源
             *        
             * 
             */
            //获取CPU 的核心数量
            int i1 = Runtime.getRuntime().availableProcessors();
            System.out.println(i1);
            ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2,
                    Runtime.getRuntime().availableProcessors(),
                    3,
                    TimeUnit.SECONDS,
                    new LinkedBlockingDeque<>(3),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.DiscardOldestPolicy()
            );
            try {
                // 最大承载:Deque + max
                //超出最大承载 RejectedExecutionException
                for (int i = 0; i < 9; i++) {
                    threadPool.execute(()->{
                        System.out.println(Thread.currentThread().getName()+"ok");
                    });
                }
            } finally {
                threadPool.shutdown();
            }
        }
    }

# 11. 四大函数式接口&Stream #

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

@FunctionalInterface
    public interface Runnable {
        /**
         * When an object implementing interface <code>Runnable</code> is used
         * to create a thread, starting the thread causes the object's
         * <code>run</code> method to be called in that separately executing
         * thread.
         * <p>
         * The general contract of the method <code>run</code> is that it may
         * take any action whatsoever.
         *
         * @see     java.lang.Thread#run()
         */
        public abstract void run();
    }
    // 超级多的FunctionalInterface
    // 简化编程模型,在新版本的框架中大量的使用

## 11.1 Function 函数式接口 ##

package func;
    
    import java.util.function.Function;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: func
     * Version: 1.0
     * Function 函数式接口
     * @author qingzhi.wu 2020/7/18 13:33
     */
    public class Dmo01 {
        public static void main(String[] args) {
            //函数式接口
            Function<String,String> function = (str)->{return str;};
        };
    }

## 11.2 Predicate 消费型接口 ##

Predicate<String> s = (o) -> {o ==null};
    Predicate<String> s = o -> o ==null;

## 11.3 Consumer 消费型接口 ##

Consumer<String> consumer = o -> System.out.println(o);

## 11.4 Supplier 供给型接口 ##

Supplier<String> s1 = ()->"aaa";

## 11.4 Stream ##

package com.bigdata.stream;
    
    import java.util.Arrays;
    import java.util.Collections;
    import java.util.List;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.stream
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 13:49
     */
    public class Demo01 {
        public static void main(String[] args) {
            /**
             * 现在有5个用户 请进行如下筛选
             * 1. id 必须是偶数
             * 2. 年龄大于23岁
             * 3. 用户名转为大写字母
             * 4. 用户名字母倒着排序
             * 5. 只输出一个用户
             */
            User u1 = new User(1,"a",21);
            User u2 = new User(2,"b",22);
            User u3 = new User(3,"c",23);
            User u4 = new User(4,"d",24);
            User u5 = new User(5,"e",25);
            // 集合就是存储
            List<User> users = Arrays.asList(u1, u2, u3, u4, u5);
            //计算交给流
            users.stream()
                    .filter(u->u.getId() %2 == 0)
                    .filter(u -> u.getAge()>23)
                    .map(u->{
                        u.setName(u.getName().toUpperCase());
                        return u;
                    })
                    .sorted((uu1,uu2)->{ return uu1.getName().compareTo(uu2.getName());})
    
                    .forEach(System.out::println);
        }
    }

## 12.ForkJoin ##

![20210318102130147.gif][]

package com.bigdata.forkjoin;
    
    import java.util.concurrent.RecursiveAction;
    import java.util.concurrent.RecursiveTask;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.forkjoin
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 15:38
     */
    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 {
                long middle = (start + end) / 2;
                ForkJoinDemo task1 = new ForkJoinDemo(start, middle);
                ForkJoinDemo task2 = new ForkJoinDemo(start, middle);
                task1.fork(); //拆分任务, 把任务压入线程队列
                task2.fork(); //拆分任务, 把任务压入线程队列
    
                return task1.join() + task2.join();
            }
        }
    }

package com.bigdata.forkjoin;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    import java.util.stream.LongStream;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.forkjoin
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 16:01
     */
    public class Test {
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            test3();
        }
    
        // 普通
        public static void test1(){
            long start = System.currentTimeMillis();
            Long sum = 0L;
            for (long i = 1L; i < 10_0000_0000L; i++) {
                sum+=i;
            }
    
            long end = System.currentTimeMillis();
            System.out.println("sum=" +"时间" +(end-start)/1000.0 +"秒");
        }
        public static void test2() throws ExecutionException, InterruptedException {
            long start = System.currentTimeMillis();
            ForkJoinPool forkJoinPool = new ForkJoinPool();
            ForkJoinDemo task = new ForkJoinDemo(1L,10_0000_0000L);
            ForkJoinTask<Long> submit = forkJoinPool.submit(task);
            System.out.println(submit.get());
            //Long aLong = task.get();
            long end = System.currentTimeMillis();
            System.out.println("sum=" +"时间" +(end-start));
        }
        public static void test3(){
            long start = System.currentTimeMillis();
            //Stream 并行流
    
            long reduce = LongStream.range(1L, 10_0000_0000L).parallel()
                    .reduce(0, Long::sum);
            System.out.println(reduce);
            long end = System.currentTimeMillis();
            System.out.println("sum=" +"时间" +(end-start));
        }
    }

# 12. 异步回调 #

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

package com.bigdata.future;
    
    import java.util.concurrent.CompletableFuture;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.TimeUnit;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.future
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 16:18
     * <p>
     * 异步调用: CompletableFuture
     * 成功回调
     * 失败回调
     */
    public class Demo01 {
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            //没有返回值的 runAsync 异步回调
    //        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
    //
    //            try {
    //                TimeUnit.SECONDS.sleep(2);
    //            } catch (InterruptedException e) {
    //                e.printStackTrace();
    //            }
    //
    //            System.out.println(Thread.currentThread().getName() + "runAsync");
    //        });
    //
    //        System.out.println(11111);
    //        completableFuture.get();
    //      有返回值的异步回调
            CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
                int i = 10/0;
                        return 1024;
                    }
            );
    
            completableFuture.whenComplete((t, u) -> {
                        System.out.println("t>" + t);
                        System.out.println("u>" + u);
    
                    }
    
            ).exceptionally((e)->{
                System.out.println(e.getMessage());
                return 233;
            })
            ;
        }
    }

# 13. JMM #

> 请你谈谈 你对 volatile 的 理解

volatile 是Java虚拟机提供 `轻量级的同步机制`

1.  保证可见性
2.  不保证原子性
3.  禁止指令重排

> 什么是 JMM

JMM : Java内存模型,不存在的东西,概念! 约定!

**关于JMM的一些同步的约定**

1.  线程解锁前,必须把共享变量`立刻`刷回主存
2.  线程加锁前,必须读取主存中的最新值到工作内存中
3.  加锁和解锁是同一把锁

**8 种操作**

![20210318102130148.gif][]

*  lock（锁定）：作用于主内存的变量，把一个变量标识为一条线程独占状态。
 *  unlock（解锁）：作用于主内存变量，把一个处于锁定状态的变量释放出来，释放后的变量才可以被其他线程锁定。
 *  read（读取）：作用于主内存变量，把一个变量值从主内存传输到线程的工作内存中，以便随后的 load 动作使用
 *  load（载入）：作用于工作内存的变量，把 read 操作从主内存中得到的变量值放入工作内存的变量副本中。
 *  use（使用）：作用于工作内存的变量，把工作内存中的一个变量值传递给执行引擎，每当虚拟机遇到一个需要使用变量的值的字节码指令时将会执行这个操作。
 *  assign（赋值）：作用于工作内存的变量，它把一个从执行引擎接收到的值赋值给工作内存的变量，每当虚拟机遇到一个给变量赋值的字节码指令时执行这个操作。
 *  store（存储）：作用于工作内存的变量，把工作内存中的一个变量的值传送到主内存中，以便随后的 write 的操作。
 *  write（写入）：作用于主内存的变量，它把 store 操作从工作内存中一个变量的值传送到主内存的变量中。

如果要把一个变量从主内存中复制到工作内存，就需要按顺序地执行 read 和 load 操作，如果把变量从工作内存同步回主内存中，就要按顺序地执行 store 和 write 操作。Java 内存模型只要求上述操作必须按顺序执行，而没有保证必须是连续执行。也就是 read 和 load 之间，store 和 write 之间是可以插入其他指令的，如对主内存中的变量 a、b 进行访问时，可能的顺序是 read a，read b，load b， load a。Java 内存模型还规定了在执行上述八种基本操作时，必须满足如下规则：

1 不允许 read 和 load、store 和write 操作之一单独出现

2 不允许一个线程丢弃它的最近 assign 的操作，即变量在工作内存中改变了之后必须同步到主内存中。

3 不允许一个线程无原因地（没有发生过任何 assign 操作）把数据从工作内存同步回主内存中。

4 一个新的变量只能在主内存中诞生，不允许在工作内存中直接使用一个未被初始化（load 或 assign）的变量。即对一个变量实施 use 和 store 操作之前，必须先执行过了 assign 和 load 操作。

5 一个变量在同一时刻只允许一条线程对其进行 lock 操作，lock 和 unlock 必须成对出现

6 如果对一个变量执行 lock 操作，将会清空工作内存中此变量的值，在执行引擎使用这个变量前需要重新执行 load 或 assign 操作初始化变量的值

7 如果一个变量事先没有被 lock 操作锁定，则不允许对它执行 unlock 操作；也不允许去 unlock 一个被其他线程锁定的变量

8 对一个变量执行 unlock 操作之前，必须先把此变量同步到主内存中（执行 store 和 write 操作）

package com.bigdata.tvolatile;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.tvolatile
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 16:57
     */
    public class JMMDemo {
        private static int num =0;
        public static void main(String[] args) {
            //main 线程
            new Thread(()->{
                while (num ==0){
    
                }
            }).start();
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
             
            num =1;
            System.out.println(num);
    
        }
    }

![20210318102130150.gif][]

# 14. Volatile可见性及非原子性验证 #

## 14.1 可见性 ##

package com.bigdata.tvolatile;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.tvolatile
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 16:57
     */
    public class JMMDemo {
        private volatile static int num =0;
        public static void main(String[] args) {
            //main 线程
            new Thread(()->{
                while (num ==0){
    
                }
            }).start();
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
            num =1;
            System.out.println(num);
    
        }
    }

![20210318102130153.gif][]

## 14.2 不保证原子性 ##

**原子性: 不可分割**

线程A在执行任务的时候,不能被打扰的,也不能被分割,要么同时成功要么同时失败

package com.bigdata.tvolatile;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.tvolatile
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 17:05
     */
    public class VDemo02 {
        //volatile 不保证原子性
        private volatile static int num= 0;
    
        //理论上 num的结果应该为 2万
        public static void main(String[] args) {
            for (int i = 0; i < 20; i++) {
                new Thread(()->{
                    for (int j = 0; j < 1000; j++) {
                        add();
                    }
                }).start();
            }
    
            while (Thread.activeCount()>2){
                Thread.yield();
            }
    
            System.out.println(Thread.currentThread().getName()+"==="+num);
        }
    
        public  static void add(){
            num++;
        }
    }

> 如果不加lock 和 synchronized,怎么保证原子性?

![20210318102130187.gif][]

我们的 `++`操作并不是原子性操作

我们可以使用原子类进行修改

private volatile static AtomicInteger num= new AtomicInteger();
    
        public  static void add(){
            num.getAndIncrement(); //CAS 
        }
    
    Unsafe类是一个很特殊的存在!

## 14.3 禁止指令重排 ##

> 指令重排

指令重排的意思就是在我们的写完的代码之后,在编译器进行将代码编译的时候会出于一些优化,调整我们代码的执行顺序

> 内存屏障

1.  保证特定的操作的执行顺序
2.  可以保证某些变量的内存可见性

# 15.单例设计模式 #

package com.bigdata.singleton;
    
    /**
     * Copyright (c) 2019 leyou ALL Rights Reserved Project: learning Package: com.bigdata.singleton
     * Version: 1.0
     *
     * @author qingzhi.wu
     * @date 2020/6/15 20:11
     */
    public class LazySingleton {
    
      private static volatile LazySingleton instance;
    
      private LazySingleton() {
        synchronized (LazySingleton.class){
          if (instance != null){
            throw  new RuntimeException("不要试图使用反射破坏单例");
          }
        }
    
      }
    
      public static synchronized LazySingleton getInstance() {
    
        if (instance == null) {
          synchronized (LazySingleton.class) {
            if (instance == null) {
              instance = new LazySingleton();
            }
          }
        }
        return instance;
      }
    }

# 16. CAS #

## 16.1 什么是CAS ##

CAS : 比较当前工作内存中的值和主内存中的值,如果这个值是期望的,那么则执行操作,如果不是就一直循环

缺点:

1.  循环会耗时
2.  一次性只能保证一个共享变量的原子性
3.  ABA问题

## 16.2 ABA问题 ##

![20210318102130190.gif][]

package com.bigdata.cas;
    
    import java.util.concurrent.atomic.AtomicInteger;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.cas
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 18:09
     */
    public class CASDemo {
        public static void main(String[] args) {
            //比较并交换
            //public final boolean compareAndSet(int expect, int update) {
            // 如果我期望的值达到了,那么就更新,否则就不更新 CAS 是CPU 的并发原语
            AtomicInteger atomicInteger = new AtomicInteger(2020);
    
            atomicInteger.compareAndSet(2020,2021);
    
            atomicInteger.compareAndSet(2021,2020);
    
            atomicInteger.compareAndSet(2020,2021);
            System.out.println(atomicInteger);
    
    
        }
    }

## 16.3 原子引用解决ABA问题 ##

**其实就是乐观锁的思想**

package com.bigdata.cas;
    
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    import java.util.concurrent.atomic.AtomicStampedReference;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.cas
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 18:09
     */
    public class CASDemo {
        static final  AtomicStampedReference<Integer> atomicStampedReference = new AtomicStampedReference<>(1,1);
    
        public static void main(String[] args) {
            //比较并交换
            //public final boolean compareAndSet(int expect, int update) {
            // 如果我期望的值达到了,那么就更新,否则就不更新 CAS 是CPU 的并发原语
            //final AtomicInteger atomicInteger = new AtomicInteger(2020);
    
            new Thread(()->{
                int stamp = atomicStampedReference.getStamp();
                System.out.println("a1=>"+atomicStampedReference.getStamp());
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                atomicStampedReference.compareAndSet(1,2,atomicStampedReference.getStamp(),atomicStampedReference.getStamp()+1);
                System.out.println("a2=>"+atomicStampedReference.getStamp());
                atomicStampedReference.compareAndSet(2,1,atomicStampedReference.getStamp(),atomicStampedReference.getStamp()+1);
                System.out.println("a3=>"+atomicStampedReference.getStamp());
    
            },"A").start();
    
            new Thread(()->{
                int stamp = atomicStampedReference.getStamp();
                System.out.println("b1=>"+atomicStampedReference.getStamp());
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                atomicStampedReference.compareAndSet(1,2,atomicStampedReference.getStamp(),atomicStampedReference.getStamp()+1);
                System.out.println("b1=>"+atomicStampedReference.getStamp());
    
    
            },"A").start();
    
    
    
        }
    }

# 17. 各种锁 #

## 17.1 公平锁,非公平锁 ##

公平锁: 非常公平,不能插队,必须先来后到

ReentrantLock lock = new ReentrantLock(true);

非公平锁: 非常不公平,可以插队

ReentrantLock lock = new ReentrantLock();

## 17.2 可重入锁 ##

**可重入锁(递归锁)**

![20210318102130151.gif][]

package com.bigdata.lock;
    
    /**
     * Copyright (c) 2019 bigdata ALL Rights Reserved
     * Project: learning
     * Package: com.bigdata.lock
     * Version: 1.0
     *
     * @author qingzhi.wu 2020/7/18 18:44
     */
    public class Demo01 {
        public static void main(String[] args) {
            Phone phone = new Phone();
            new Thread(()->{
                phone.sms();
            },"A").start();
    
            new Thread(()->{
                phone.sms();
            },"B").start();
    
        }
    
    }
    
    class Phone {
        public synchronized void sms(){
            System.out.println(Thread.currentThread().getName()+"sms");
            call();
        }
        public synchronized void call(){
            System.out.println(Thread.currentThread().getName()+"call");
    
        }
    }

package com.bigdata.lock;
    
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
    * Copyright (c) 2019 bigdata ALL Rights Reserved
    * Project: learning
    * Package: com.bigdata.lock
    * Version: 1.0
    *
    * @author qingzhi.wu 2020/7/18 18:44
    */
    public class Demo02 {
       public static void main(String[] args) {
           Phone phone = new Phone();
           new Thread(()->{
               phone.sms();
          },"A").start();
    
           new Thread(()->{
               phone.sms();
          },"B").start();
    
      }
    
    }
    
    class Phone2 {
       Lock lock = new ReentrantLock();
       public synchronized void sms(){
           lock.lock();
           // lock 锁必须配对,不然就死锁
           try {
               System.out.println(Thread.currentThread().getName()+"sms");
               call();
          } finally {
               lock.unlock();
          }
      }
       public synchronized void call(){
           lock.lock();
           try {
               System.out.println(Thread.currentThread().getName()+"call");
          } finally {
               lock.unlock();
          }
    
      }
    }

## 17.3 自旋锁 ##

![20210318102130184.gif][]

package com.bigdata.lock;
    
    import java.util.concurrent.atomic.AtomicReference;
    
    /**
    * Copyright (c) 2019 bigdata ALL Rights Reserved
    * Project: learning
    * Package: com.bigdata.lock
    * Version: 1.0
    *
    * @author qingzhi.wu 2020/7/18 18:53
    */
    public class MyLockDemo {
       static final  AtomicReference<Thread> atomicReference = new AtomicReference<>();
    
    
       public void lock(){
           Thread thread = Thread.currentThread();
           System.out.println(Thread.currentThread().getName() +"-->lock");
           while (atomicReference.compareAndSet(null,thread)){
    
          }
      }
       public void unlock(){
           Thread thread = Thread.currentThread();
           System.out.println(Thread.currentThread().getName() +"-->unlock");
           atomicReference.compareAndSet(thread,null);
      }
    }

## 17.4 死锁 ##

![20210318102130147.gif][]

package com.bigdata.demo05;
    
    /**
    * Copyright (c) 2019 bigdata ALL Rights Reserved
    * Project: learning
    * Package: com.bigdata.demo05
    * Version: 1.0
    * 死锁: 多个线程互相抱着对方需要的资源,然后形成僵持
    * @author qingzhi.wu 2020/7/15 21:34
    */
    public class DeadLock {
       public static void main(String[] args) {
           Mackeup g1 = new Mackeup(0,"灰姑凉");
           Mackeup g2 = new Mackeup(1,"白雪公主");
           g1.start();
           g2.start();
    
      }
    }
    
    //口红
    class Lipstick{
    
    }
    //镜子
    class Mirror{
    
    }
    class Mackeup extends Thread{
       static Lipstick lipstick = new Lipstick();
       static Mirror mirror = new Mirror();
    
       int choice;//选择
       String girlName;//使用化妆品的人
    
       public Mackeup(int choice, String girlName) {
           this.choice = choice;
           this.girlName = girlName;
      }
    
       @Override
       public void run() {
           //化妆
           try {
               makeup();
          } catch (InterruptedException e) {
               e.printStackTrace();
          }
      }
       private void makeup() throws InterruptedException {
           if (choice==0){
               synchronized (lipstick){//获得口红的锁
                   System.out.println(this.girlName +"获得口红的锁");
                   Thread.sleep(1000);
                   synchronized (mirror){
                       System.out.println(this.girlName +"获得镜子的锁");
    
                  }           }
    
          }else {
               synchronized (mirror){//获得口红的锁
                   System.out.println(this.girlName +"获得镜子的锁");
                   Thread.sleep(2000);
                   synchronized (lipstick){
                       System.out.println(this.girlName +"获得口红的锁");
                  }
    
              }
    
          }
      }
    }

**死锁的排查**

1.  使用`jps`定位进程号

![图片][2221b1ddcc4d9d1f66c632d311e9b8b8.png]

1.  使用`jstack`可以定位死锁

![图片][73cab43918b7b5aca665a48cf7bc21d1.png]

[93271ca937744b8aa9b48d491e604f6a.png]: /images/20221022/6d5ae0ed146a42089552519f3fc56994.png
[20210318102130147.gif]: /images/20221022/58b30cab40d6491a85de8692c72ca140.png
[20210318102130185.gif]: /images/20221022/de777344fd524adfa247bfbdf87b557e.png
[20210318102130183.gif]: /images/20221022/128d5e60469c47de932fb47a43ac3459.png
[cabb912869bd7d59aa2938acffde0852.png]: /images/20221022/21895ab40a924981ae3f0b3328c28f53.png
[20210318102130152.gif]: /images/20221022/cd5fb54a3b1043c3ac1bfb6bb36b3c5c.png
[20210318102130188.gif]: /images/20221022/0acaa5897c4e43529eaec95fe19c5e1c.png
[20210318102130149.gif]: /images/20221022/e8bb2ac876534d5f92ef03e2f6b5b2ac.png
[20210318102130184.gif]: /images/20221022/7284d0b7fae84c41968e8a307bc03db2.png
[20210318102130148.gif]: /images/20221022/9322836eeb424b208cb33fc1d9ca9a01.png
[20210318102130150.gif]: /images/20221022/5df7a349ff4b41f2bccbff66b91dcd02.png
[20210318102130153.gif]: /images/20221022/0e2546652d134df198d90e68c7d0ddae.png
[20210318102130187.gif]: /images/20221022/588ea8cc8bde4c7cb2072987700865f6.png
[20210318102130190.gif]: /images/20221022/e24757e818114f63b1eff59acdb52ee7.png
[20210318102130151.gif]: /images/20221022/c01117f1fc15472da6d9d9985020eeed.png
[2221b1ddcc4d9d1f66c632d311e9b8b8.png]: /images/20221022/5f8d941d7a2a444eb20a036c870d760d.png
[73cab43918b7b5aca665a48cf7bc21d1.png]: /images/20221022/e3926d5d847a426da0cb7769b33ba52c.png