并发编程进阶，深入浅出理解 JUC

JUC 多线程进阶

1.什么是 JUC

源码 + 官方文档

JUC 是 java util concurrent

业务：普通的线程代码 Thread

Runnable：** 没有返回值、效率相比于Callable 相对较低！**

2.线程和进程

线程与进程的关系

进程是一个程序的执行；一个进程可以包含多个线程，线程是程序执行的最小单位（资源调度的最小单位）

java 默认有几个线程？

2 个，main与GC

线程：开了一个进程 Typore，写字，自动保存（线程负责的，输入）

对于 Java 而言：Thread、Runnable、Callable

Java 可以开启线程吗？

答：不能，调用 Thread.start()方法实质上是调用的本地方法(native 修饰)，底层是 C++ 来实现的，因为 Java 无法操作硬件。

    public synchronized void start() {
         /**
          * This method is not invoked for the main method thread or "system"
          * group threads created/set up by the VM. Any new functionality added
          * to this method in the future may have to also be added to the VM.
          *
          * A zero status value corresponds to state "NEW".
          */
         if (threadStatus != 0)
             throw new IllegalThreadStateException();
 
         /* Notify the group that this thread is about to be started
          * so that it can be added to the group's list of threads
          * and the group's unstarted count can be decremented. */
         group.add(this);
 
         boolean started = false;
         try {
             start0();
             started = true;
         } finally {
             try {
                 if (!started) {
                     group.threadStartFailed(this);
                 }
             } catch (Throwable ignore) {
                 /* do nothing. If start0 threw a Throwable then
                   it will be passed up the call stack */
             }
         }
     }
     //本地方法，底层C++，Java无法直接操作硬件
     private native void start0();

并发、并行

并发（多线程操作同一个资源）

• CPU 一核，模拟出来多条线程。快速交替

并行（多个人一起干事）

• CPU 多核，多个线程可以同时执行。

 package cn.fyyice.juc;
 
 public class Test1 {
     public static void main(String[] args) {
         //获取CPU核数
         System.out.println(Runtime.getRuntime().availableProcessors());
     }
 }
 

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

线程有几个状态

     public enum State {
         //  新生
         NEW,
 
         //  运行
         RUNNABLE,
 
         //  阻塞
         BLOCKED,
 
         //  等待
         WAITING,
 
         //  超时等待
         TIMED_WAITING,
 
         //  终止
         TERMINATED;
     }

wait 与 sleep 的区别

**1、**来自不同的类

wait => Object

sleep => Thread

**2、**关于锁的释放

wait：会释放锁

sleep：睡觉了，抱着锁睡觉，不会释放！

**3、**使用的范围是不同的

wait：必须在同步代码块中

sleep：可以在任何地方睡

**4、**是否需要捕获异常

wait：不需要捕获异常

sleep：必须要捕获异常（超市等待）

在线程 sleep 过后，会让出 CPU 资源。目的是不让当前线程独自霸占该进程所获的 CPU 资源，以留一定时间给其他线程执行的机会。所以不会占用 cpu。时间到了会正常返回，线程处于就绪状态，然后参与 cpu 调度，获取到 cpu 资源之后就可以运行。

3.Lock 锁

在真正的多线程开发中（公司），线程就是一个单独的资源类，没有任何附属的操作。

这里以卖票的 demo 来进行讲解

传统 synchronized 解决

本质：排队

 package cn.fyyice.juc;
 
 public class SaleTicket {
     public static void main(String[] args) {
         //并发：多线程操作一个资源类
         Ticket t = new Ticket();
 
         //@FunctionalInterface函数式接口，jdk 1.8 lombda表达式 (参数)->{代码块}
         new Thread(()->{ for (int i = 0; i < 50; i++) t.sale();},"甲").start();
         new Thread(()->{ for (int i = 0; i < 50; i++) t.sale();},"乙").start();
         new Thread(()->{ for (int i = 0; i < 50; i++) t.sale();},"丙").start();
     }
 }
 
 // OOP资源类
 class Ticket{
     private Integer ticket = 20;
 
     public synchronized void sale(){
         if (ticket > 0){
             System.out.println(Thread.currentThread().getName()+"卖出了编号:"+ticket--+"的票,剩余:"+ticket);
         }
     }
 }

Lock 接口

 Lock l = ...; l.lock(); try { // access the resource protected by this lock } finally { l.unlock(); } 

当在不同范围内发生锁定和解锁时，必须注意确保在锁定时执行的所有代码由 try-finally 或 try-catch 保护，以确保在必要时释放锁定。

Lock 实现提供了使用 synchronized 方法和语句的附加功能，通过提供非阻塞尝试来获取锁（ tryLock() ），尝试获取可被中断的锁（ lockInterruptibly()） ，以及尝试获取可以超时（ tryLock(long, TimeUnit) ）。

LOCK 中声明的所有方法：

实现类**：**

所有已知实现类：

1. ReentrantLock** （可重入锁---常用）**
2. ReentrantReadWriteLock.ReadLock** （读锁）**
3. ReentrantReadWriteLock.WriteLock（写锁）

ReentrantLock 源码：

  /**
      * Creates an instance of {@code ReentrantLock}.
      * This is equivalent to using {@code ReentrantLock(false)}.
      */
     public ReentrantLock() {
         sync = new NonfairSync();
     }
 
     /**
      * Creates an instance of {@code ReentrantLock} with the
      * given fairness policy.
      *
      * @param fair {@code true} if this lock should use a fair ordering policy
      */
     public ReentrantLock(boolean fair) {
         sync = fair ? new FairSync() : new NonfairSync();
     }

NonfairSync：非公平锁，可以插队

FairSync：公平锁，先来后到

 package cn.fyyice.juc;
 
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
 public class SaleTicket2 {
     public static void main(String[] args) {
         //并发：多线程操作一个资源类
         Ticket2 t = new Ticket2();
 
         //@FunctionalInterface函数式接口，jdk 1.8 lombda表达式 (参数)->{代码块}
         new Thread(()->{ for (int i = 0; i < 30; i++) t.sale();},"A").start();
         new Thread(()->{ for (int i = 0; i < 30; i++) t.sale();},"B").start();
         new Thread(()->{ for (int i = 0; i < 30; i++) t.sale();},"C").start();
     }
 }
 
 // OOP资源类
 class Ticket2{
     private Integer ticket = 20;
     Lock lock = new ReentrantLock();
     public void sale(){
         //加锁
         lock.lock();
         try {
             if (ticket > 0){
                 System.out.println(Thread.currentThread().getName()+"卖出了编号:"+ticket--+"的票,剩余:"+ticket);
             }
         }catch (Exception e){
             e.printStackTrace();
         }finally {
             //解锁
             lock.unlock();
         }
     }
 }

synchronized 与 Lock 区别

1. Synchronized 是内置的 Java 关键字， Lock 是一个 Java 类
2. Synchronized 无法判断获取锁的状态，Lock 可以判断是否获取到了锁
3. **Synchronized 会自动释放锁， Lock 必须要手动释放锁！如果不释放锁，**死锁
4. Synchronized 线程 1（获得锁，但发生了阻塞），线程 2 会一直等待；Lock 锁就不一定会等待下去（tryLock 方法，尝试获取锁）
5. Synchronized 可重入锁，不可以中断的，非公平；Lock 可重入锁。可以判断锁，自己设定公平/非公平
6. Synchronized 适合锁少量的代码同步问题；Lock 适合锁大量的同步代码

锁是什么，如何判断锁的是谁

4.生产者和消费者问题

 线程之间的通信问题：生产者和消费者问题。    等待换新，通知唤醒
 线程交替执行  A   B 操作统一变量    num = 0
 A   num+1
 B   num-1

面试的：

单例模式、排序算法、生产者和消费者问题、死锁

package cn.fyyice.juc.pc;

public class A {
    public static void main(String[] args) {
        Data data = new Data();
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"A").start();
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"B").start();
    }
}

//判断等待、业务、通知
class Data{
    private int number = 0;

    public synchronized void increment() throws InterruptedException {
        if (number != 0){
            //等待
            this.wait();
        }
        number ++;
        System.out.println(Thread.currentThread().getName()+"-------->"+number);
        //通知 业务完成
        this.notifyAll();
    }

    public synchronized void decrement() throws InterruptedException {
        if (number == 0){
            //等待
            this.wait();
        }
        number --;
        System.out.println(Thread.currentThread().getName()+"-------->"+number);
        //通知 业务完成
        this.notifyAll();
    }
}

问题存在，A B C D 四个线程。虚假唤醒！

新加入线程后，是否安全？

** **不安全。用 if 判断的话，唤醒后线程会从 wait 之后的代码开始运行，但是不会重新判断 if 条件，直接继续运行 if 代码块之后的代码，而如果使用 while 的话，也会从 wait 之后的代码运行，但是唤醒后会重新判断循环条件，如果不成立再执行 while 代码块之后的代码块，成立的话继续 wait。

官方文档解释：

线程也可以唤醒，而不会被通知，中断或超时**，即所谓的**=虚假唤醒=** 。 虽然这在实践中很少会发生，但应用程序必须通过测试应该使线程被唤醒的条件来防范，并且如果条件不满足则继续等待。 换句话说，等待应该总是出现在循环中，就像这样：**

  synchronized (obj) {
         while (<condition does not hold>)
             obj.wait(timeout);
         ... // Perform action appropriate to condition
     } 

解决方法：

****if 改为 white！****防止虚假唤醒

package cn.fyyice.juc.pc;

public class A {
    public static void main(String[] args) {
        Data data = new Data();
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"A").start();
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"B").start();
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"C").start();
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"D").start();

    }
}

//判断等待、业务、通知
class Data{
    private int number = 0;

    public synchronized void increment() throws InterruptedException {
        while (number != 0){
            //等待
            this.wait();
        }
        number ++;
        System.out.println(Thread.currentThread().getName()+"-------->"+number);
        //通知 业务完成
        this.notifyAll();
    }

    public synchronized void decrement() throws InterruptedException {
        while (number == 0){
            //等待
            this.wait();
        }
        number --;
        System.out.println(Thread.currentThread().getName()+"-------->"+number);
        //通知 业务完成
        this.notifyAll();
    }
}

JUC 版本的生产者和消费者问题

传统的：Synchronized wait notifyAll/notify

新版：Lock await signal

官方文档：

public interface Condition

Condition 因素出 Object 监视器方法（ wait ， notify和notifyAll ）成不同的对象，以得到具有多个等待集的每个对象，通过将它们与使用任意的组合的效果Lock 个**实现。 **Lock 替换 synchronized**方法和语句的使用， **Condition 取代了对象监视器方法的使用。

条件（也称为条件队列或条件变量** ）为一个线程暂停执行（“等待”）提供了一种方法，直到另一个线程通知某些状态现在可能为真。 因为访问此共享状态信息发生在不同的线程中，所以它必须被保护，因此某种形式的锁与该条件相关联。 等待条件的关键属性是它原子地释放相关的锁并挂起当前线程，就像**Object.wait** 。**

class BoundedBuffer {
   final Lock lock = new ReentrantLock();
   final Condition notFull  = lock.newCondition(); 
   final Condition notEmpty = lock.newCondition(); 

   final Object[] items = new Object[100];
   int putptr, takeptr, count;

   public void put(Object x) throws InterruptedException {
     lock.lock(); try {
       while (count == items.length)
         notFull.await();
       items[putptr] = x;
       if (++putptr == items.length) putptr = 0;
       ++count;
       notEmpty.signal();
     } finally { lock.unlock(); }
   }

   public Object take() throws InterruptedException {
     lock.lock(); try {
       while (count == 0)
         notEmpty.await();
       Object x = items[takeptr];
       if (++takeptr == items.length) takeptr = 0;
       --count;
       notFull.signal();
       return x;
     } finally { lock.unlock(); }
   }
 } 

代码实现：

package cn.fyyice.juc.pc;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class B {
    public static void main(String[] args) {
        Data2 data = new Data2();
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"A1").start();
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"B1").start();
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.increment();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"C1").start();
        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                try {
                    data.decrement();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"D1").start();

    }
}

//判断等待、业务、通知
class Data2{
    private int number = 0;

    Lock lock = new ReentrantLock();
    Condition condition = lock.newCondition();

    public void increment() throws InterruptedException {
        lock.lock();
        try {
            while (number != 0){
                //等待
                condition.await();
            }
            number ++;
            System.out.println(Thread.currentThread().getName()+"-------->"+number);
            //通知 业务完成
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public void decrement() throws InterruptedException {
        lock.lock();
        try {
            while (number == 0){
                //等待
                condition.await();
            }
            number --;
            System.out.println(Thread.currentThread().getName()+"-------->"+number);
            //通知 业务完成
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

任何一个新的技术绝对不仅仅只是覆盖了原来的技术

Condition 的优势

精准的通知和操作线程

1. 设定标志位

   Condition condition1 = lock.newCondition();
   Condition condition2 = lock.newCondition();
   Condition condition3 = lock.newCondition();
   
   //	通过对监控的Condition进行signal方法进行唤醒
   

2. 设定多个监视器 Condition

通过 while 循环对标志位进行判断。满足条件，则对指定的监视器执行 singal 操作，从而达到精确唤醒

5.8 锁现象

8 锁，就是关于锁的八个问题

package cn.fyyice.juc.lock8;

import java.util.concurrent.TimeUnit;

public class Test {
    public static void main(String[] args) {
        Phone phone = new Phone();

        new Thread(()->{
            phone.send();

        },"A").start();

        try {
            System.out.println("我要等待3s");
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        new Thread(()->{
            phone.call();
        },"B").start();
    }
}

class Phone {
    // synchronized锁的对象是方法的调用者！
    // 这里两个方法用的是同一个锁，谁先拿到谁先执行！
    public synchronized void send(){
        try {
            System.out.println("我是phone中的等待");
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("发信息");
    }
    public synchronized void call(){
        System.out.println("打电话");
    }
}

package cn.fyyice.juc.lock8;

import java.util.concurrent.TimeUnit;

public class Test2 {
    public static void main(String[] args) {
        Phone2 phone = new Phone2();
        Phone2 phone2 = new Phone2();

        new Thread(()->{
            phone.send();

        },"A2").start();

        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        new Thread(()->{
            phone2.call();
        },"B2").start();
    }
}

class Phone2 {
    // synchronized锁的对象是方法的调用者！
    // 这里两个方法用的是同一个锁，谁先拿到谁先执行！
    public synchronized void send(){
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName()+"-->发信息");
    }
    public synchronized void call(){
        System.out.println(Thread.currentThread().getName()+"-->打电话");
    }

    public void hello(){
        System.out.println("hello");
    }
}

package cn.fyyice.juc.lock8;

import java.util.concurrent.TimeUnit;

public class Test3 {
    public static void main(String[] args) {
        Phone3 phone = new Phone3();

        new Thread(() -> {
            phone.send();

        }, "A").start();

        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        new Thread(() -> {
            phone.call();
        },"B").start();

    }
}

class Phone3 {
    // synchronized锁的对象是方法的调用者！
    // 这里两个方法用的是同一个锁，谁先拿到谁先执行！
    public static synchronized void send(){
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName()+"-->发信息");
    }
    public static synchronized void call(){
        System.out.println(Thread.currentThread().getName()+"-->打电话");
    }
}

package cn.fyyice.juc.lock8;

import java.util.concurrent.TimeUnit;

public class Test4 {
    public static void main(String[] args) {
        Phone4 phone = new Phone4();

        new Thread(() -> {
            phone.send();

        }, "A").start();

        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        new Thread(() -> {
            phone.call();
        },"B").start();

    }
}

class Phone4 {
    // synchronized锁的对象是方法的调用者！
    // 这里两个方法用的是同一个锁，谁先拿到谁先执行！
    public static synchronized void send(){
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName()+"-->发信息");
    }
    public synchronized void call(){
        System.out.println(Thread.currentThread().getName()+"-->打电话");
    }
}

顺序：发信息在前，打电话在后

1. 标准情况下，两个线程先打印 发信息还是打电话？
2. send()方法延迟 4s，两个线程先打印发信息 还是打电话？
3. 在加入普通方法 hello 后，两个线程先打印发信息 还是 hello？
4. 两个对象，2 个同步方法， 发信息还是 打电话？
5. 增加两个静态的同步方法，只有一个对象，先打印 发短信还是打电话？
6. 两个对象，增加两个静态的同步方法，只有一个对象，先打印 发短信还是打电话？
7. static 方法的 send，同步方法 call，只有一个对象，先打印 发短信还是打电话？
8. static 方法的 send，同步方法 call，只有两个对象，先打印 发短信还是打电话？

回复：

问题1,2：  按照顺序执行，原因是因为这两个方法都是加了锁的，而synchronized是修饰在方法上面的，所以锁的是这个方法的调用对象，在这里两个方法都是同一个调用对象，所以谁先拿到锁谁先执行。
问题3：	不是同步的方法，不考虑锁的因素。这里其实还是应该先执行发信息，只不过因为sleep了，而hello不需要去等待这个锁，所以先执行的hello。
问题4：
问题5：	先打印发短信，再打印打电话。因为在加入static修饰后，变成了静态方法，在类初始化的时候就会执行，相当于类一加载就有了！锁的是Class对象（模板）
问题6：	先打印发短信，再打印打电话。因为锁的是Class对象，类模板是同一个，所以谁先拿到锁谁先执行，这里就是顺序执行。
问题7：	先打印打电话，再打印发信息。因为锁的对象不一样，发信息方法锁的对象是Class对象，而打电话方法锁的是类调用对象。锁不一样，不需要去等待锁的释放。
问题8：	同上。

小结：

• new this 具体的一个类对象
• static Class 唯一的一个模板
• 在主线程与子线程中的代码执行顺序问题
  1. 当主线程代码段在(一个或多个)子线程中间时，首先执行的还是主线程代码段。因为刚开始时，只有主线程在使用 CPU 的执行权，因为其他两个线程还没有被创建，这时主线程的代码就自上而下的去执行。在线程都创建完成后，此时就存在多个线程了，而线程的执行需要抢到 CPU 资源去执行，所以在后续就是谁先抢到 CPU 资源谁先执行了。

6.集合类不安全

List 不安全

package cn.fyyice.juc.unfairy;

import java.sql.Connection;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;

//java.util.ConcurrentModificationException 并发修改异常
public class ListTest1 {
    public static void main(String[] args) {
        /**
         *  并发下，list是不安全的
         *  解决方案：
         *  1.List<String> list = Vector()<>集合类
         *  2.List<String> list = Collections.synchronizedList(new ArrayList<>())
         *  3.List<String> list = new CopyOnWriteArrayList<>();
         *      CopyOnWrite 写入时复制    COW 优化策略
         *      多个线程调用的时候，list读取是固定的，但是在写入的时候可能会被后面的线程给覆盖掉
         *      在写入的时候避免覆盖，造成数据问题
         *      读写分离
         */
        List<String> list = new CopyOnWriteArrayList<>();

        for (int i = 1; i <= 10; i++) {
            new Thread(()->{
                list.add(UUID.randomUUID().toString().substring(0,6));
                System.out.println(list);
            },String.valueOf(i)).start();

        }
    }
}

Set 不安全

package cn.fyyice.juc.unfairy;

import java.util.HashSet;
import java.util.Set;
import java.util.UUID;

public class SetList {
    public static void main(String[] args) {
        //不安全
        Set<Object> set = new HashSet<>();
        /**
         * 解决方案：
         *  Collections.synchronizedSet(new HashSet<>());
         *  new CopyOnWriteArraySet<>()
         */
        for (int i = 0; i < 30; i++) {
            new Thread(()->{
                set.add(UUID.randomUUID().toString().substring(0,6));
            },String.valueOf(i)).start();
        }
    }
}

hashSet 底层是什么？

public HashSet() {
    map = new HashMap<>();
}
// add  set 本质就是 map key 是无法重复的！
public boolean add(E e) {
    return map.put(e, PRESENT)==null;
}

private static final Object PRESENT = new Object();	//不变的值

Map 不安全

    /**
     * The default initial capacity - MUST be a power of two.
     */
    static final int DEFAULT_INITIAL_CAPACITY = 1 << 4; // aka 16 默认容量

    /**
     * The maximum capacity, used if a higher value is implicitly specified
     * by either of the constructors with arguments.
     * MUST be a power of two <= 1<<30.
     */
    static final int MAXIMUM_CAPACITY = 1 << 30;

    /**
     * The load factor used when none specified in constructor.
     */
    static final float DEFAULT_LOAD_FACTOR = 0.75f;	//加载因子

package cn.fyyice.juc.unfairy;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

public class MapTest {
    public static void main(String[] args) {
        //map是这样用的？默认等价于什么？
        Map<String, String> map = new HashMap<>();
        //加载因子，初始化容量
        /**
         *  new ConcurrentHashMap<>();
         *  Collections.synchronizedMap(new HashMap<>());
         */
        for (int i = 1; i < 20; i++) {
            new Thread(()->{
                map.put(Thread.currentThread().getName(), UUID.randomUUID().toString().substring(0,5));
                System.out.println(map);
            },String.valueOf(i)).start();
        }
    }
}

7.Callable

@FunctionalInterface
public interface Callable<V>

返回结果并可能引发异常的任务。 实现者定义一个没有参数的单一方法，称为 call** 。**

Callable 接口类似于Runnable** ，因为它们都是为其实例可能由另一个线程执行的类设计的。 然而，A **Runnable 不返回结果，也不能抛出被检查的异常。

该Executors类包含的实用方法，从其他普通形式转换为 Callable 类。

1. 有返回值
2. 可以跑出异常
3. 方法不同，run()/call()
4. 通过 FutureTask 来实现

• 一个 FutureTask 可以用来包装Callable或Runnable对象。因为 FutureTask 实现 Runnable，一 FutureTask 可以提交执行Executor**。**** **
• 

    /**
     * Creates a {@code FutureTask} that will, upon running, execute the
     * given {@code Callable}.
     *
     * @param  callable the callable task
     * @throws NullPointerException if the callable is null
     */
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }

    /**
     * Creates a {@code FutureTask} that will, upon running, execute the
     * given {@code Runnable}, and arrange that {@code get} will return the
     * given result on successful completion.
     *
     * @param runnable the runnable task
     * @param result the result to return on successful completion. If
     * you don't need a particular result, consider using
     * constructions of the form:
     * {@code Future<?> f = new FutureTask<Void>(runnable, null)}
     * @throws NullPointerException if the runnable is null
     */
    public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

代码实现

package cn.fyyice.juc.callable;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class CallableTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // new Thread(new Runnable() {}).start();
        // new Thread(new FutureTask<V>()).start());

        MyThread thread = new MyThread();
        //适配类
        FutureTask<String> task = new FutureTask<>(thread);
        new Thread(task,"A").start();
        new Thread(task,"C").start();
        new Thread(task,"D").start();
        //这个get方法可能会出现阻塞！把他放到最后、或者使用异步通信来处理
        System.out.println(task.get());
    }
}

class MyThread implements Callable<String> {

    @Override
    public String call() throws Exception {
        System.out.println("成功进入Callable方法");
        return Thread.currentThread().getName()+"--->hello";
    }
}

输出结果：

成功进入Callable方法
A--->hello

Process finished with exit code 0

注：

• 有缓存
• 结果可能需要等待，会阻塞。

8.常用辅助类

• CountDownLatch(减法计数器)

****官方解释：****CountDownLatch 是一个同步的辅助类，允许一个或多个线程，等待其他一组线程完成操作，再继续执行。

可以理解为倒计时锁**。就比如玩 LOL 匹配/排位。一场对局的开始必须等待所有人加载 100% 才能开始，所以就算你是外星人，**是小霸王，你也得等着。

demo 01

package cn.fyyice.juc.utils;

import java.util.concurrent.CountDownLatch;

//计数器
public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        //总数是10，必须要执行任务的时候才使用
        CountDownLatch count = new CountDownLatch(10);
        for (int i = 0; i < 10; i++) {
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+"溜了");
                count.countDown();
            },String.valueOf(i)).start();
        }
        //等待计数器归0，再向下执行
        count.await();
        System.out.println("遛完了，关门");
    }
}

demo 02

package cn.fyyice;

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

public class CountDown {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(5);
        for (int i = 0; i < 5; i++) {
            new Thread(()->{
                try{
                    int time  = new Random().nextInt(4)+1;
                    System.out.println(Thread.currentThread().getName()+"--->start loading,need "+time+" s");
                    TimeUnit.SECONDS.sleep(time);
                    System.out.println(Thread.currentThread().getName()+"--->loading finally!");
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    countDownLatch.countDown();
                }
            },"Player " + i).start();
        }

        try {
            System.out.println("主线程开始执行...");
            countDownLatch.await();
            System.out.println("主线程执行结束...");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

原理：

=countDown();=** **** **** 数量减 1**

=await()=** ****等待计数器归零，然后再向下执行**

当计数器归零时，=await()=**方法就会被唤醒，继续执行。**

• CyclicBarrier（加法计数器）

****官方解释：****CyclicBarrier 是一个同步的辅助类，允许一组线程相互之间等待，达到一个共同点，再继续执行。

可以把 CyclicBarrier 看成是个障碍，所有的线程必须到齐后才能一起通过这个障碍。常用于多线程分组计算。比如一个大型的任务，常常需要分配好多子任务去执行，只有当所有子任务都执行完成时候，才能执行主任务；通俗点可以理解为一个公会副本奖励，需要公会所有人成员都完成至少在线一小时才能发布奖励。

// 计数，并执行一个线程
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

package cn.fyyice.juc.utils;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierDemo {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{
            System.out.println("太下饭了，饱了");
        });

        for (int i = 1; i <= 7; i++) {
            //因为i是临时变量
            final int temp = i;
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+"-->第"+temp+"次吃饭");
                try {
                    //  每执行一次await() 内部会执行一次+1
                    cyclicBarrier.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            },String.valueOf(i)).start();

        }
    }
}

• Semaphore

 public Semaphore(int permits) {
        sync = new NonfairSync(permits);
}

计数信号量**。从概念上讲，一个信号量维护一组允许。每个 **acquire()**块如果必要的许可证前，是可用的，然后把它。每个 **release()**添加许可，潜在收购方释放阻塞。然而，不使用实际允许的对象； **Semaphore**只是计数的数量和相应的行为。 **

信号量通常是用来限制线程的数量比可以访问一些（物理或逻辑）资源。例如，这里是一个类使用一个信号量来控制访问的项目池：

package cn.fyyice.juc.utils;

import java.util.Random;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

public class SemaphoreDemo {
    public static void main(String[] args) {
        //限流
        Semaphore semaphore = new Semaphore(4);
        for (int i = 0; i < 10; i++) {
            new Thread(()->{
                try {
                    //得到
                    semaphore.acquire();
                    int time  = new Random().nextInt(4)+1;
                    System.out.println(Thread.currentThread().getName()+"得到了车位,并且要停"+time+"s");
                    TimeUnit.SECONDS.sleep(time);
                    System.out.println(Thread.currentThread().getName()+"开走了，空出一个车位");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    // 释放
                    semaphore.release();
                }
            },String.valueOf(i)).start();
        }

    }
}

原理：

=semaphore.acquire();=** ****获得，假设已经满了，那么就会等待，直到被释放位置！**

=semaphore.release();=** ****释放，会将当前的信号量释放 +1，然后唤醒等待的线程！**

作用：多个共享资源互斥作用！并发限流，控制最大的线程数！

9.读写锁

ReadWriteLock

**所有已知实现类： **

ReentrantReadWriteLock

public interface ReadWriteLock

**一个 **ReadWriteLock**保持一对一联系 locks，只读操作和书写。的 **read lock**可能被多个线程同时举行的读者，只要没有作家。 **write lock**是独家的。 **

读的时候可以被多线程同时读，写的时候只能由一个线程去写。

• 读写锁允许访问共享数据比用互斥锁允许更多的并发级别。它利用的事实，而只有一个单一的线程在一个时间（一个作家线程）可以修改共享数据，在许多情况下，任何数量的线程可以同时读取数据（因此读者线程）。在理论上，并发的读写锁的使用允许的增加将导致在一个互斥锁的使用性能的改进。在实践中，这种增加的并发性只会完全实现一个多处理器，然后，只有当共享数据的访问模式是合适的。

package cn.fyyice.juc.raw;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 又称为   独占锁（写锁）    一次只能被一个线程占有
 *          共享锁（读锁）    多个线程可以同时占有
 * ReadWriteLock
 * 读-读  可以共存
 * 读-写  不能共存
 * 写-写  不能共存
 */
public class ReadWriteLockDemo {
    public static void main(String[] args) {
//        MyCache myCache = new MyCache();
        MyLockCache myCache = new MyLockCache();
        for (int i = 1; i <= 5; i++) {
            final int temp = i;
            new Thread(()->{
                myCache.put(temp+"", UUID.randomUUID().toString().substring(0,5));
            },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 MyLockCache{

    private volatile Map<String,Object> map = new HashMap<>();
    //读写锁：更加细粒度的控制
    private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

    //在写操作的时候希望同一时间，只有一个线程进行写操作，避免覆盖
    public void put(String key,Object value){
        readWriteLock.writeLock().lock();
        try {
            System.out.println(Thread.currentThread().getName()+"写入："+value);
            map.put(key,value);
            System.out.println(Thread.currentThread().getName()+"写入完成");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            readWriteLock.writeLock().unlock();
        }
    }

    //读的时候加锁是为了防止   在读的时候发生写操作
    public void get(String key){
        readWriteLock.readLock().lock();
        try {
            System.out.println(Thread.currentThread().getName()+"读取："+key);
            Object obj = map.get(key);
            System.out.println("----------"+Thread.currentThread().getName()+"读取完成---------");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            readWriteLock.readLock().unlock();
        }
    }

}

/**
 * 自定义缓存
 */
class MyCache{

    private volatile Map<String,Object> map = new HashMap<>();

    public void put(String key,Object value){
        System.out.println(Thread.currentThread().getName()+"写入："+value);
        map.put(key,value);
        System.out.println("----------"+Thread.currentThread().getName()+"写入完成---------");
    }

    public void get(String key){
        System.out.println(Thread.currentThread().getName()+"读取："+key);
        Object obj = map.get(key);
        System.out.println("----------"+Thread.currentThread().getName()+"读取完成---------");
    }
}

10.阻塞队列

• 阻塞队列（FIFO）

  • 写入：如果队列满了，就必须阻塞等待（取出）
  • 取：如果队列是空的，必须阻塞等待生产（写入）

  InInterface BlockingQueue****

  • • 参数类型 **
      E -元素举行此集合中的类型 **
    • All Superinterfaces: **
      Collection ， Iterable ， Queue **
    • All Known Subinterfaces: **
      BlockingDeque ， TransferQueue **
    • 所有已知实现类： **
      ArrayBlockingQueue， DelayQueue， LinkedBlockingDeque， LinkedBlockingQueue， LinkedTransferQueue， PriorityBlockingQueue， SynchronousQueue **

  ---

  什么情况下会使用阻塞队列：

  • 多线程并发处理
  • 线程池

  

四组 API

BlockingQueue<Object> blockingQueue = new ArrayBlockingQueue<>(3);

SynchronousQueue 同步队列

没有容量

进去一个元素，必须等待元素取出来过后，才能 put 进去

package cn.fyyice.juc.bq;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

public class SynchronousQueueDemo {
    public static void main(String[] args) {
        // 同步队列     和其他BlockingQueue不一样，synchronousQueue不存储元素
        BlockingQueue<String> synchronousQueue = new SynchronousQueue<>();

        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName()+"--->put1");
                synchronousQueue.put("1");
                System.out.println(Thread.currentThread().getName()+"--->put2");
                synchronousQueue.put("2");
                System.out.println(Thread.currentThread().getName()+"--->put3");
                synchronousQueue.put("3");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"A").start();

        new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+"--->get1");
                synchronousQueue.take();
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+"--->get2");
                synchronousQueue.take();
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+"--->get3");
                synchronousQueue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"B").start();
    }
}

11.线程池

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

池化技术

程序的运行，本质：占用系统资源。我们需要做的是优化资源的使用！

常见：线程池、连接池、对象池

池化技术**：事先准备好一些资源，有人要用就直接来这里拿，用完过后回收。**

线程池的好处：

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

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

三大方法

1. Executors.newSingleThreadExecutor();
2. Executors.newFixedThreadPool(线程数量);
3. Executors.newCachedThreadPool();

package cn.fyyice.juc.pool;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

//Executor  工具类，3大方法
public class Demo1 {
    public static void main(String[] args) {
        //创建单个线程池
        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
        //创建一个固定大小的线程池
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(6);
        //创建一个可伸缩的线程池
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();


        try {
            for (int i = 0; i < 10; i++) {
                //使用了线程池之后，使用线程池来创建线程

                //始终只有一个线程执行
//                singleThreadExecutor.execute(()->{
//                    System.out.println(Thread.currentThread().getName()+"----> start");
//                });

                //固定线程数量执行
//                fixedThreadPool.execute(()->{
//                    System.out.println(Thread.currentThread().getName()+"----> start");
//                });

                //根据cpu支持线程数量执行
                cachedThreadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName()+"----> start");
                });

            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //线程池用完，程序结束，关闭线程池
            //singleThreadExecutor.shutdown();
            //fixedThreadPool.shutdown();
            cachedThreadPool.shutdown();
        }
    }
}

七大参数

源码分析

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;
}

手动创建线程池

package cn.fyyice.juc.pool;

import java.util.concurrent.*;

public class Demo2 {
    public static void main(String[] args) {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 6, 3, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(3),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
        System.out.println("当前CPU核数:"+Runtime.getRuntime().availableProcessors());
        try {
            //最大承载 maxImumPoolSize + capacity
            for (int i = 1; i <= 8; i++) {
                threadPoolExecutor.execute(()->{
                    System.out.println(Thread.currentThread().getName()+"---> deal");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPoolExecutor.shutdown();
        }
    }
}

四种拒绝策略

1. new ThreadPoolExecutor.AbortPolicy()
   

   如果超出了最大处理量（maxImumPoolSize + capacity），则不进行处理，抛出异常。

2. new ThreadPoolExecutor.CallerRunsPolicy()
   

   哪来的去哪里。

3. new ThreadPoolExecutor.DiscardOldestPolicy(由调用线程处理该任务)
   

   （放弃最旧的策略）队列满了，尝试去和最早的竞争。也不会抛出异常

4. new ThreadPoolExecutor.DiscardPolicy()
   

   队列满了，直接丢掉任务，不会抛出异常。（解决不了问题，就解决提出问题的人）

小结

最大线程倒地如何定义？（调优）

1. CPU 密集型
   • 看电脑配置，几核就是定义几，可以保持 CPU 型效率最高
2. IO 密集
   • 判断你的程序中，十分耗 IO 的线程，然后设置大于这个数的线程

12.四大函数式接口

作用：简化底层编程模型

• lambda 表达式
• 链式编程
• 函数式接口
• Stream 流式计算

函数式接口

只有只一个方法的接口

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

四大函数式接口

• Consumer
• Function
• Predicate
• Supplier

代码测试

1.Function

@FunctionalInterface
public interface Function<T, R> {

    /**
     * Applies this function to the given argument.
     *
     * @param t the function argument
     * @return the function result
     */
    R apply(T t);

}

package cn.fyyice.juc.function;

import java.util.function.Function;

public class FunctionDemo {
    public static void main(String[] args) {
//			可用作工具类
//        Function<String, String> f = new Function<String, String>() {
//            @Override
//            public String apply(String str) {
//                return str;
//            }
//        };
        Function<String, String> f = (str)-> {return str;};
        System.out.println(f.apply("hello world"));
    }
}

2.Predicate

package cn.fyyice.juc.function;

import java.util.function.Predicate;

/**
 * 断定性接口，返回值只能是boolean
 */
public class PredicateDemo {
    public static void main(String[] args) {
//        Predicate<String> predicate = new Predicate<String>() {
//            @Override
//            public boolean test(String str) {
//                return str.isEmpty();
//            }
//        };
        Predicate<String> p = str->{return str.isEmpty();};
        System.out.println(p.test("123"));

    }
}

3.Consumer

@FunctionalInterface
public interface Consumer<T> {

    /**
     * Performs this operation on the given argument.
     *
     * @param t the input argument
     */
    void accept(T t);

package cn.fyyice.juc.function;

import java.util.function.Consumer;

/**
 * 消费性接口：只有输入，没有返回值
 */
public class ConsumerDemo {
    public static void main(String[] args) {

        Consumer<String> consumer = str->{
            System.out.println(str);
        };
        consumer.accept("hello");
    }
}

4.Supplier

@FunctionalInterface
public interface Supplier<T> {

    /**
     * Gets a result.
     *
     * @return a result
     */
    T get();
}

package cn.fyyice.juc.function;

import java.util.function.Supplier;

/**
 * 供给型接口：没有参数，只有返回值
 */
public class SupplierDemo {
    public static void main(String[] args) {
        Supplier supplier = ()->{return 1024;};
        System.out.println(supplier.get());
    }
}