JUC

1. 什么是JUC

JUC就是java.util .concurrent工具包的简称。

业务: 普通的线程代码, 之前都是用的thread或者runnable接口

但是相比于callable来说,thread没有返回值,且效率没有callable高。

2. 线程和进程

进程：程序的集合。

一个进程可以包含多个线程，至少包含一个。

Java默认有main线程和GC线程。

对于Java而言，三种开启线程的方式。Thread、Runnable、Callable。

Java只能通过本地方法调用底层C++开启线程，Java无法直接操作硬件。

2.1. 并发、并行

并发编程：并发、并行

并发：多线程操作同一个资源。交替进行。快速交替产生并行的假象。

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

并行：同时进行。

CPU多核，多个线程同时执行。线程池。

public class Test1 {
    public static void main(String[] args) {
        // 获取CPU的核数
        // CPU密集型，IO密集型
        System.out.println(Runtime.getRuntime().availableProcessors());
    }
}

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

2.2. 线程的状态

Thread.State定义了六个状态：

NEW 新生
RUNNABLE 运行
BLOCKED 阻塞
WAITING 等待
TIMED_WAITING 有限时间等待
TERMINATED 终止

2.3. wait/sleep的区别

来自不同的类。wait是Object类，sleep是Thread类。

import java.util.concurrent.TimeUnit;
public class Test1 {
    public static void main(String[] args) throws InterruptedException {
        TimeUnit.DAYS.sleep(1);
    }
}

wait会释放锁，sleep不会释放锁，即抱着锁睡觉。
使用范围不同。
1. wait，notify和notifyAll只能在同步控制方法或者同步控制块里面使用。
2. sleep可以在任何地方执行。
相同：都需要捕获异常。InterruptedException

3. Lock锁

在java的锁的实现上就有共享锁和独占锁的区别，而这些实现都是基于AbstractQueuedSynchronizer对于共享同步和独占同步的支持。

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

3.1. 锁的独占与共享

java并发包提供的加锁模式分为独占锁和共享锁，独占锁模式下，每次只能有一个线程能持有锁，ReentrantLock就是以独占方式实现的互斥锁。共享锁，则允许多个线程同时获取锁，并发访问共享资源，如：ReadWriteLock。AQS的内部类Node定义了两个常量SHARED（共享）和EXCLUSIVE（独占），他们分别标识 AQS队列中等待线程的锁获取模式。

很显然，独占锁是一种悲观保守的加锁策略，它避免了读/写冲突，如果某个只读线程获取锁，则其他读线程都只能等待，这种情况下就限制了不必要的并发性，因为读操作并不会影响数据的一致性。共享锁则是一种乐观锁，它放宽了加锁策略，允许多个执行读操作的线程同时访问共享资源。 java 的并发包中提供了ReadWriteLock，读-写锁。它允许一个资源可以被多个读操作访问，或者被一个写操作访问，但两者不能同时进行。

共享模式和独占模式的区别在于，独占模式只允许一个线程获得资源，而共享模式允许多个线程获得资源。

3.2. 锁的公平与非公平

锁的公平与非公平，是指线程请求获取锁的过程中，是否允许插队。在公平锁上，线程将按他们发出请求的顺序来获得锁；而非公平锁则允许在线程发出请求后立即尝试获取锁，如果可用则可直接获取锁，尝试失败才进行排队等待。ReentrantLock提供了两种锁获取方式，FairSyn和NofairSync。

公平锁：十分公平，可以先来后到。

非公平锁：十分不公平，可以插队。（默认）

public ReentrantLock() {
  sync = new NonfairSync();
}

public ReentrantLock(boolean fair) {
  sync = fair ? new FairSync() : new NonfairSync();
}

结论：ReentrantLock是以独占锁的加锁策略实现的互斥锁，同时它提供了公平和非公平两种锁获取方式。

3.3. 传统synchronized

本质: 队列和锁, synchronized 放在方法上锁的是this,放在代码块中锁的是()里面的对象

package com.gs.demo01;

/**
 * 基本的卖票例子
 *
 * @author admin
 * @date 2021/8/25 8:29 下午
 */
public class SaleTicketDemo01 {
    public static void main(String[] args) {
        // 并发：多线程操作同一个资源类，把资源类丢入线程
        Ticket ticket = new Ticket();

        // @FunctionalInterface 函数式接口
        new Thread(()->{
            for (int i = 1; i < 40; i++) {
                ticket.sale();
            }
        },"A").start();
        new Thread(()->{
            for (int i = 1; i < 40; i++) {
                ticket.sale();
            }
        },"B").start();
        new Thread(()->{
            for (int i = 1; i < 40; i++) {
                ticket.sale();
            }
        },"C").start();

    }
}

/**
 * 资源类 OOP编程
 */
class Ticket {
    // 属性、方法
    private int number = 50;

    // 卖票的方式
    public synchronized void sale() {
        if (number > 0) {
            System.out.println(Thread.currentThread().getName() + "卖出了第" + number-- + "张票，剩余" + number + "张票");
        }
    }

}

3.4. Lock接口

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

三个实现类

ReentrantLock 可重入锁（常用），

ReentrantReadWriteLock.ReadLock 读锁，

ReentrantReadWriteLock.WriteLock 写锁

package com.gs.demo01;

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

/**
 * 基本的卖票例子
 *
 * @author admin
 * @date 2021/8/25 8:29 下午
 */
public class SaleTicketDemo02 {
    public static void main(String[] args) {
        // 并发：多线程操作同一个资源类，把资源类丢入线程
        Ticket2 ticket = new Ticket2();

        new Thread(()->{
            for (int i = 1; i < 40; i++) {
                ticket.sale();
            }
        },"A").start();
        new Thread(()->{
            for (int i = 1; i < 40; i++) {
                ticket.sale();
            }
        },"B").start();
        new Thread(()->{
            for (int i = 1; i < 40; i++) {
                ticket.sale();
            }
        },"C").start();

    }
}

/**
 * 资源类 OOP编程
 */

/**
 * Lock三部曲
 * 1. new ReentrantLock();
 * 2. lock.lock();    // 加锁
 * 3. finally => lock.unlock();  // 解锁
 */
class Ticket2 {
    // 属性、方法
    private int number = 50;

    Lock lock = new ReentrantLock();

    // 卖票的方式
    public synchronized void sale() {

        lock.lock();    // 加锁

        try {

            // 业务代码
            if (number > 0) {
                System.out.println(Thread.currentThread().getName() + "卖出了第" + number-- + "张票，剩余" + number + "张票");
            }

        } finally {
            lock.unlock();  // 解锁
        }
    }
}

synchronized和lock锁的区别

synchronized内置的java关键字,Lock是一个java类
synchronized无法判断获取锁的状态, Lock可以判断是否获取到了锁
synchronized会自动释放锁,Lock必须要手动释放锁!如果不是释放锁,会产生死锁，ReentrantLock 和 synchronized 不一样，需要手动释放锁，所以使用 ReentrantLock的时候一定要手动释放锁，并且加锁次数和释放次数要一样
0synchronized 线程1(获得锁,阻塞),线程2(等待); Lock锁就不一定会等待下去
synchronized 可重入锁,不可以中断的,非公平的; Lock锁,可重入的,可以判断锁,非公平(可自己设置为公平锁);
synchronized 适合锁少量的代码同步问题,Lock 适合锁大量的同步代码

3.5. 生产者消费者问题

Synchronized版本

旧版本：Synchronized wait notify

package com.gs.demo01.pc;

/**
 * 线程之间的通信问题，生产者和消费者问题
 *
 * 等待唤醒，通知唤醒
 *
 * 线程交替执行 A B操作同一个变量 num = 0
 * A num + 1
 * B num - 1
 * @author admin
 * @date 2021/8/25 11:29 下午
 */
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;

    // +1
    public synchronized void increment() throws InterruptedException {
        if (number != 0) {
            // 等待
            this.wait();
        }
        number++;
        System.out.println(Thread.currentThread().getName() + "=>" + number);
        // 通知其他线程
        this.notifyAll();

    }

    // - 1
    public synchronized void decrement() throws InterruptedException {
        if (number == 0) {
            // 等待
            this.wait();
        }
        number--;
        System.out.println(Thread.currentThread().getName() + "=>" + number);
        // 通知其他线程
        this.notifyAll();
    }
}

上述代码的问题：如果是两个线程以上，线程还安全吗？

不安全，因为if代码块中的wait()会释放锁，由于其他线程notifyAll唤醒的所有线程会在wait()位置继续，所以会导致重复加。

在多个线程通信时，必须使用 while 判断防止虚假唤醒。if判断改为while判断。

因为if只会执行一次，线程醒来会接着向下执行if（）外边的

而while不会，会重新判断条件是否满足，满足才会向下执行while（）外边的

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

synchronized (obj) { while () obj.wait(timeout); ... // Perform action appropriate to condition }

package com.gs.demo01.pc;

/**
 * 线程之间的通信问题，生产者和消费者问题
 *
 * 等待唤醒，通知唤醒
 * @author admin
 * @date 2021/8/25 11:29 下午
 */
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;

    // +1
    public synchronized void increment() throws InterruptedException {
        while (number != 0) {
            // 等待
            this.wait();
        }
        number++;
        System.out.println(Thread.currentThread().getName() + "=>" + number);
        // 通知其他线程
        this.notifyAll();

    }

    // - 1
    public synchronized void decrement() throws InterruptedException {
        while (number == 0) {
            // 等待
            this.wait();
        }
        number--;
        System.out.println(Thread.currentThread().getName() + "=>" + number);
        // 通知其他线程
        this.notifyAll();
    }
}

JUC版本

通过Lock 找到 Condition（监视者）

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

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 com.gs.demo01.pc;

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

/**
 * 线程之间的通信问题，生产者和消费者问题
 *
 * 等待唤醒，通知唤醒
 * @author admin
 * @date 2021/8/25 11:29 下午
 */
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();
                }
            }
        },"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 Data2 {

    private int number = 0;

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

    // +1
    public void increment() throws InterruptedException {
        lock.lock();
        try {
            while (number != 0) {
                // 等待
                condition.await();
            }
            number++;
            System.out.println(Thread.currentThread().getName() + "=>" + number);
            // 通知其他线程
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    // - 1
    public void decrement() throws InterruptedException {
        lock.lock();
        try {
            while (number == 0) {
                // 等待
                condition.await();
            }
            number--;
            System.out.println(Thread.currentThread().getName() + "=>" + number);
            // 通知其他线程

            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

上述代码的问题：随机的状态

如何有序执行？

Condition精准通知和唤醒线程

有序唤醒

package com.gs.demo01.pc;

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

/**
 * A执行完调用B，B执行完调用C，C执行完调用A
 * @author admin
 * @date 2021/8/25 11:29 下午
 */
public class C {
    public static void main(String[] args) {
        Data3 data = new Data3();

        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                data.printA();
            }
        },"A").start();

        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                data.printB();
            }
        },"B").start();

        new Thread(()->{
            for (int i = 0; i < 10; i++) {
                data.printC();
            }
        },"C").start();


    }
}

/**
 * 数字 资源类
 *
 * 判断等待 业务 通知
 */
class Data3 {

    private int number = 1; // 1A 2B 3C

    private Lock lock = new ReentrantLock();
    private Condition condition1 = lock.newCondition();
    private Condition condition2 = lock.newCondition();
    private Condition condition3 = lock.newCondition();

    public void printA() {
        lock.lock();
        try {
            while (number != 1) {
                // 等待
                condition1.await();
            }
            // 业务：判断，执行，通知
            System.out.println(Thread.currentThread().getName() + " AAAAA");
            number = 2;
            condition2.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public void printB() {
        lock.lock();
        try {
            while (number != 2) {
                condition2.await();
            }
            // 业务：判断，执行，通知
            System.out.println(Thread.currentThread().getName() + " BBBBB");
            number = 3;
            condition3.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public void printC() {
        lock.lock();
        try {
            while (number != 3) {
                condition3.await();
            }
            // 业务：判断，执行，通知
            System.out.println(Thread.currentThread().getName() + " CCCCC");
            number = 1;
            condition1.signal();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

3.6. 八锁现象（关于锁的八个问题）

锁的对象有具体的实例、Class。

package com.gs.lock8;

import java.util.concurrent.TimeUnit;

/**
 * 1. 标准情况下，Astart后延时1s，再startB，两个线程先打印哪个？
 * 先发短信。
 *
 * 2. 上述情况下，sendSMS延迟4s，先打印哪个？
 * 先发短信。
 *
 * 上述问题的原因：
 * synchronized锁的对象是方法的调用者，即phone。
 * 两个方法用的其实是同一个锁，谁先拿到锁谁会被先执行。
 *
 *
 * @author admin
 * @date 2021/8/26 9:12 上午
 */
public class Test1 {
    public static void main(String[] args) throws InterruptedException {
        Phone phone = new Phone();

        new Thread(() -> {
            try {
                phone.sendSMS();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "A"
        ).start();

        // JUC下的休眠
        TimeUnit.SECONDS.sleep(1);

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

class Phone {

    public synchronized void sendSMS() throws InterruptedException {
        TimeUnit.SECONDS.sleep(4);
        System.out.println("sendSMS");
    }

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

package com.gs.lock8;

import java.util.concurrent.TimeUnit;

/**
 * 3. 增加一个普通方法后，先执行发短信还是先执行hello
 * 先hello。
 * hello方法没有锁，不是同步方法，不受锁的影响
 *
 *
 * @author admin
 * @date 2021/8/26 9:12 上午
 */
public class Test2 {
    public static void main(String[] args) throws InterruptedException {
        Phone2 phone = new Phone2();

        new Thread(() -> {
            phone.sendSMS();
        }, "A"
        ).start();

        // JUC下的休眠
        TimeUnit.SECONDS.sleep(1);

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

class Phone2 {

    public synchronized void sendSMS() {
        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");
    }
}

package com.gs.lock8;

import java.util.concurrent.TimeUnit;

/**
 * 4. 两个对象，两个同步方法，先执行哪一个？
 * 打电话
 * 对象不同，锁不住。
 * 发短信的线程休眠了4s，所以打电话先打印。
 *
 * @author admin
 * @date 2021/8/26 9:12 上午
 */
public class Test2 {
    public static void main(String[] args) throws InterruptedException {
        Phone2 phone1 = new Phone2();
        Phone2 phone2 = new Phone2();

        new Thread(() -> {
            phone1.sendSMS();
        }, "A"
        ).start();

        // JUC下的休眠
        TimeUnit.SECONDS.sleep(1);

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

class Phone2 {

    public synchronized void sendSMS() {
        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");
    }
}

package com.gs.lock8;

import java.util.concurrent.TimeUnit;

/**
 * 5. 增加两个静态的同步方法，先打印发短信还是打电话
 * 发短信
 *
 * synchronized 锁的对象是方法的调用者
 * static 静态方法 类一加载就有了
 * 上述二者一块用表示锁的是Class模板
 * Phone3唯一的一个Class对象 ==> Phone3.Class
 * 所以两个方法用的是同一把锁
 *
 *
 * 6. 两个对象，两个同步方法，先发短信还是打电话
 * 发短信
 *
 *
 * @author admin
 * @date 2021/8/26 9:12 上午
 */
public class Test3 {
    public static void main(String[] args) throws InterruptedException {
        Phone3 phone = new Phone3();
        Phone3 phone1 = new Phone3();

        new Thread(() -> {
            phone.sendSMS();
        }, "A"
        ).start();

        // JUC下的休眠
        TimeUnit.SECONDS.sleep(1);

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

class Phone3 {


    public static synchronized void sendSMS() {
        try {
            TimeUnit.SECONDS.sleep(4);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("sendSMS");
    }

    public static synchronized void call() {
        System.out.println("call");
    }
}

package com.gs.lock8;

import java.util.concurrent.TimeUnit;

/**
 * 7. 一个静态同步方法，一个普通同步方法，先打印哪个？
 * 打电话
 *
 * 静态同步方法 锁的是Class类模板
 * 普通同步方法 锁的是调用者
 * @author admin
 * @date 2021/8/26 9:12 上午
 */
public class Test4 {
    public static void main(String[] args) throws InterruptedException {
        Phone4 phone = new Phone4();

        new Thread(() -> {
            phone.sendSMS();
        }, "A"
        ).start();

        // JUC下的休眠
        TimeUnit.SECONDS.sleep(1);

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

class Phone4 {

    // 静态同步方法 锁的是Class类模板
    public static synchronized void sendSMS() {
        try {
            TimeUnit.SECONDS.sleep(4);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("sendSMS");
    }

    // 普通同步方法 锁的是调用者
    public static void call() {
        System.out.println("call");
    }
}

package com.gs.lock8;

import java.util.concurrent.TimeUnit;

/**
 * 8. 一个静态同步方法，一个普通同步方法，两个对象分别调用，先打印哪个？
 * 打电话
 *
 * 静态同步方法 锁的是Class类模板
 * 普通同步方法 锁的是调用者
 * @author admin
 * @date 2021/8/26 9:12 上午
 */
public class Test4 {
    public static void main(String[] args) throws InterruptedException {
        Phone4 phone1 = new Phone4();
        Phone4 phone2 = new Phone4();

        new Thread(() -> {
            phone1.sendSMS();
        }, "A"
        ).start();

        // JUC下的休眠
        TimeUnit.SECONDS.sleep(1);

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

class Phone4 {

    public static synchronized void sendSMS() {
        try {
            TimeUnit.SECONDS.sleep(4);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("sendSMS");
    }

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

小结：

new this 具体的对象

static Class 唯一的模板

4. 集合类不安全

4.1. List不安全

package com.gs.unsafe;

import com.gs.demo01.pc.C;
import javafx.collections.ArrayChangeListener;

import javax.xml.ws.soap.Addressing;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @author admin
 * @date 2021/8/26 10:41 上午
 */
public class ListTest {
    public static void main(String[] args) {

        // 单线程下安全
//        List<String> list = Arrays.asList("1", "2", "3");
//        list.forEach(System.out::println);

        // 并发下ArrayList不安全
        // java.util.ConcurrentModificationException 并发修改异常!
//        List<String> list = new ArrayList<>();
//        for (int i = 0; i < 10; i++) {
//            new Thread(()->{
//                list.add(UUID.randomUUID().toString().substring(0, 5));
//                System.out.println(list);
//            }, String.valueOf(i)).start();
//        }


        /*
        解决方案1：Vector()
        List<String> list = new Vector<>();
         public synchronized boolean add(E e) {...}
         */
//        List<String> list = new Vector<>();
//        for (int i = 0; i < 10; i++) {
//            new Thread(()->{
//                list.add(UUID.randomUUID().toString().substring(0, 5));
//                System.out.println(list);
//            }, String.valueOf(i)).start();
//        }

        /*
        解决方案2：使用Collections工具类下的synchronizedXXX方法对List进行封装
         public static <T> List<T> synchronizedList(List<T> list) {
        return (list instanceof RandomAccess ?
                new SynchronizedRandomAccessList<>(list) :
                new SynchronizedList<>(list));
    }
         */
//        List<String> list = Collections.synchronizedList(new ArrayList<>());
//        for (int i = 0; i < 10; i++) {
//            new Thread(() -> {
//                list.add(UUID.randomUUID().toString().substring(0, 5));
//                System.out.println(list);
//            }, String.valueOf(i)).start();
//        }

        /*
        解决方案3：new CopyOnWriteArrayList<>();

        private transient volatile Object[] array;

     public boolean add(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            Object[] elements = getArray();
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len + 1);
            newElements[len] = e;
            setArray(newElements);
            return true;
            } finally {
                lock.unlock();
            }
        }
         */
        /*
        CopyOnWrite 写入时复制 COW 计算机程序设计领域的一种优化策略
        多个线程调用的时候，list读取的时候是固定的，写入的时候可能覆盖
        使用COW在写入的时候会拷贝给调用者，写入后插入，避免覆盖造成数据问题
         */

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

    }
}

CopyOnWriteArrayList()与Vector()相比好在哪里？

Vector的add使用synchronized关键字，效率比较低。CopyOnWriteArrayList用ReentrantLock锁，效率较高。

4.2. Set不安全

package com.gs.unsafe;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 *
 * @author admin
 * @date 2021/8/26 12:25 下午
 */
public class SetTest {
    public static void main(String[] args) {

        // 与上述ListTest类的内容类似 ConcurrentModificationException
//        Set<String> set = new HashSet<>();
//        for (int i = 0; i < 30; i++) {
//            new Thread(() -> {
//                set.add(UUID.randomUUID().toString().substring(0, 5));
//                System.out.println(set);
//            }, String.valueOf(i)).start();
//        }

        /*
        解决方案1：工具类
         */
//        Set<String> set = Collections.synchronizedSet(new HashSet<>());
//        for (int i = 0; i < 30; i++) {
//            new Thread(() -> {
//                set.add(UUID.randomUUID().toString().substring(0, 5));
//                System.out.println(set);
//            }, String.valueOf(i)).start();
//        }

        /*
        解决方案2：new CopyOnWriteArraySet<>();
         */
        Set<String> set = new CopyOnWriteArraySet<>();
        for (int i = 0; i < 30; i++) {
            new Thread(() -> {
                set.add(UUID.randomUUID().toString().substring(0, 5));
                System.out.println(set);
            }, String.valueOf(i)).start();
        }
    }
}

4.3. Map不安全

package com.gs.unsafe;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author admin
 * @date 2021/8/26 12:39 下午
 */
public class MapTest {
    public static void main(String[] args) {

        // ConcurrentModificationException
        // 相当于new HashMap<1<<4, 0.75f>();
//        Map<String, String> map = new HashMap<>();
//        for (int i = 0; i < 30; i++) {
//            new Thread(()->{
//                map.put(Thread.currentThread().getName(), UUID.randomUUID().toString().substring(0, 5));
//                System.out.println(map);
//            }, String.valueOf(i)).start();
//        }

        // 方案1 Collections.synchronizedMap(new HashMap<>());
//        Map<Object, Object> map = Collections.synchronizedMap(new HashMap<>());
//        for (int i = 0; i < 30; i++) {
//            new Thread(()->{
//                map.put(Thread.currentThread().getName(), UUID.randomUUID().toString().substring(0, 5));
//                System.out.println(map);
//            }, String.valueOf(i)).start();
//        }

        // 方案2 new ConcurrentHashMap<>()
        ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
        for (int i = 0; i < 30; i++) {
            new Thread(()->{
                map.put(Thread.currentThread().getName(), UUID.randomUUID().toString().substring(0, 5));
                System.out.println(map);
            }, String.valueOf(i)).start();
        }
    }
}

5. Callable

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

可以有返回值
可以抛出异常
方法不同run()/call()

package com.gs.callable;

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

/**
 * @author admin
 * @date 2021/8/26 1:01 下午
 */
public class CallableTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        MyThread myThread = new MyThread();

        // 适配类
        FutureTask<String> futureTask = new FutureTask<>(myThread);

        new Thread(futureTask, "A").start();
        new Thread(futureTask, "B").start();

        String s = futureTask.get(); // get方法可能会产生阻塞
        System.out.println(s);
    }
}

class MyThread implements Callable<String> {
    @Override
    public String call() throws Exception {
        System.out.println("call()");
        return "123";
    }
}


/*
call()
123
 */

细节：

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

6. 常用的辅助类

6.1. CountDownLatch

允许一个或多个线程等待直到其他线程中执行的一组操作完成的同步辅助。

package com.gs.add;

import java.util.concurrent.CountDownLatch;

/**
 * 计数器
 * @author admin
 * @date 2021/8/26 1:30 下午
 */
public class CountDownLatchDemo {
    public static void main(String[] args) throws InterruptedException {
        // 总数是6
        // 必须要执行的任务的时候，再使用
        CountDownLatch countDownLatch = new CountDownLatch(6);
        for (int i = 0; i < 6; i++) {
            new Thread(()->{
                System.out.println(Thread.currentThread().getName() + "go out");
                // -1
                countDownLatch.countDown();

            }, String.valueOf(i)).start();
        }

        // 等待计数器归零，再向下执行。
        countDownLatch.await();
        System.out.println("closed");


    }
}
/*
0go out
1go out
2go out
3go out
4go out
5go out
closed
 */

countDownLatch.await();等待计数器归零后再向下执行
countDownLatch.countDown();数量-1

每次有线程调用countDown()数量-1，假设计数器为0，await()就会被唤醒，继续执行。

6.2. CyclicBarrier

允许一组线程全部等待彼此达到共同屏障点的同步辅助。循环阻塞在涉及固定大小的线程方的程序中很有用，这些线程必须偶尔等待彼此。屏障被称为循环，因为它可以在等待的线程被释放之后重新使用。

package com.gs.add;

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

/**
 * @author admin
 * @date 2021/8/26 1:41 下午
 */
public class CyclicBarrierDemo {
    /*
    集齐七颗龙珠召唤神龙
     */
    public static void main(String[] args) {

        // 召唤龙珠的线程
        CyclicBarrier cyclicBarrier = new CyclicBarrier(7, () -> {
            System.out.println("召唤神龙成功");
        });

        for (int i = 1; i <= 7; i++) {
            // lambda不能操作外部变量，本质是一个类
            final int temp = i;
            new Thread(()->{
                System.out.println(Thread.currentThread().getName() + "收集" + temp + "个龙珠");
                try {
                    cyclicBarrier.await();// 没有达到7之前会被wait，不会往下执行，到达7才会执行cyclicBarrier的任务
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

6.3. Semaphore

Semaphore：信号量

一个计数信号量。在概念上，信号量维持一组许可证。如果有必要，每个acquire()都会阻塞，直到许可证可用，然后才能使用它。每个release()添加许可证，潜在地释放阻塞获取方。但是，没有使用实际的许可证对象; Semaphore只保留可用数量的计数，并相应地执行。

package com.gs.add;

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

/**
 * @author admin
 * @date 2021/8/26 1:55 下午
 */
public class SemaphoreDemo {
    public static void main(String[] args) {
        // 线程数量：停车位
        // 限流可用
        Semaphore semaphore = new Semaphore(3);

        for (int i = 0; i < 6; i++) {
            new Thread(()->{
                try {
                    // acquire() 获得
                    semaphore.acquire();

                    System.out.println(Thread.currentThread().getName() + "抢到车位");
                    TimeUnit.SECONDS.sleep(2);
                    System.out.println(Thread.currentThread().getName() + "离开车位");

                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    // release() 释放
                    semaphore.release();
                }

            }).start();
        }
    }
}

原理：

acquire() 获得，假设如果已经满了，等待，等待被释放为止。-1
release()释放，会将当前的信号量释放，唤醒等待的线程。+1

作用：

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

7. 读写锁

ReadWriteLock接口

所有已知实现类： ReentrantReadWriteLock

public interface ReadWriteLock A ReadWriteLock维护一对关联的locks ，一个用于只读操作，一个用于写入。 read lock可以由多个阅读器线程同时进行，只要没有作者。 write lock是独家的。

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

package com.gs.rw;


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

/**
 * 独占锁（写锁） 一次只能被一个线程占有
 * 共享锁（读锁） 多个线程可以同时占有
 * ReadWriteLock
 * 读-读 可以共存
 * 读-写 不能共存
 * 写-写 不能共存
 * @author admin
 * @date 2021/8/26 3:56 下午
 */
public class ReadWriteLockDemo {
    public static void main(String[] args) {
        MyCacheLock myCache = new MyCacheLock();

        for (int i = 0; i < 5; i++) {
            final int temp = i;
            new Thread(()->{
                myCache.put(temp + "", temp + "");
            },String.valueOf(i)).start();
        }

        for (int i = 0; i < 5; i++) {
            final int temp = i;
            new Thread(()->{
                myCache.get(temp + "");
            },String.valueOf(i)).start();
        }

    }
}

/**
 * 加锁的自定义缓存
 */
class MyCacheLock {
    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() + "写入" + key);
            map.put(key, value);
            System.out.println(Thread.currentThread().getName() + "写入完成");
        } finally {
            readWriteLock.writeLock().unlock();
        }

    }

    // 取，读，所有人都可以读
    public void get(String key) {
        readWriteLock.readLock().lock();
        try {
            System.out.println(Thread.currentThread().getName() + "读取" + key);
            map.get(key);
            System.out.println(Thread.currentThread().getName() + "读取完成" + key);
        } 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() + "写入" + key);
        map.put(key, value);
        System.out.println(Thread.currentThread().getName() + "写入完成");

    }

    // 取，读
    public void get(String key) {
        System.out.println(Thread.currentThread().getName() + "读取" + key);
        map.get(key);
        System.out.println(Thread.currentThread().getName() + "读取完成" + key);
    }

}

8. 阻塞队列

阻塞队列：

8.1. BlockingQueue

Interface BlockingQueue<E>

参数类型 E - 此集合中保存的元素的类型 All Superinterfaces: Collection <E>， Iterable <E>， Queue <E> All Known Subinterfaces: BlockingDeque <E>， TransferQueue <E> 所有已知实现类： ArrayBlockingQueue ， DelayQueue ， LinkedBlockingDeque ， LinkedBlockingQueue ， LinkedTransferQueue ， PriorityBlockingQueue ， SynchronousQueue同步队列

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

多线程并发处理，线程池。

队列使用：

添加、移除

四组API

抛出异常
不会抛出异常
阻塞等待
超时等待

方式	抛出异常	不会抛出异常,有返回值	阻塞等待	超时等待
添加操作	add()	offer() 供应	put()	offer(obj,int,timeunit.status)
移除操作	remove()	poll() 获得	take()	poll(int,timeunit.status)
判断队列首部	element()	peek() 偷看,偷窥

/**
     * 抛出异常
     */
public static void test1() {
  // 队列的大小
  ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

  System.out.println(blockingQueue.add("a"));
  System.out.println(blockingQueue.add("b"));
  System.out.println(blockingQueue.add("c"));

  // 查看队首的元素
  System.out.println(blockingQueue.element());

  // IllegalStateException: Queue full
  //        System.out.println(blockingQueue.add("d"));
  System.out.println(blockingQueue.remove());
  System.out.println(blockingQueue.remove());
  System.out.println(blockingQueue.remove());

  // NoSuchElementException
  //        System.out.println(blockingQueue.remove());

}

/**
     * 不抛出异常
     */
public static void test2() {
  // 队列的大小
  ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

  System.out.println(blockingQueue.offer("a"));
  System.out.println(blockingQueue.offer("b"));
  System.out.println(blockingQueue.offer("c"));

  // 查看队首的元素
  System.out.println(blockingQueue.peek());

  // false
  System.out.println(blockingQueue.offer("d"));

  System.out.println(blockingQueue.poll());
  System.out.println(blockingQueue.poll());
  System.out.println(blockingQueue.poll());

  // null
  System.out.println(blockingQueue.poll());

}

/**
     * 等待，阻塞（一直阻塞）
     */
public static void test3() throws InterruptedException {
  ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

  blockingQueue.put("a");
  blockingQueue.put("b");
  blockingQueue.put("c");

  // 如果队列已满，则一直阻塞到队列非满
  //        blockingQueue.put("d");


  System.out.println(blockingQueue.take());
  System.out.println(blockingQueue.take());
  System.out.println(blockingQueue.take());

  // 如果队列为空，则一直阻塞到队列非空
  //        System.out.println(blockingQueue.take());
}

/**
     * 等待，阻塞（等待超时）
     */
public static void test4() throws InterruptedException {
  ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

  System.out.println(blockingQueue.offer("a"));
  System.out.println(blockingQueue.offer("b"));
  System.out.println(blockingQueue.offer("c"));

  // 等待2秒，如果2秒后仍满，则超时失败，返回false
  System.out.println(blockingQueue.offer("d", 2, TimeUnit.SECONDS));

  System.out.println(blockingQueue.poll());
  System.out.println(blockingQueue.poll());
  System.out.println(blockingQueue.poll());

  // 等待2秒，如果2秒后仍为空，则超时失败，返回null
  System.out.println(blockingQueue.poll(2, TimeUnit.SECONDS));

}

8.2. SynchronousQueue同步队列

Class SynchronousQueue<E>

没有容量

进去一个元素必须等待取出来后才能再往里面放一个元素。

put、take

package com.gs.bq;

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

/**
 * 同步队列
 * 和其他的BlockingQueue不同，SynchronousQueue不存储元素
 * put了一个元素，必须从里面先take取出，否则不能再put进值
 * @author admin
 * @date 2021/8/26 5:11 下午
 */
public class SynchronousQueueDemo {
    public static void main(String[] args) {
        SynchronousQueue<String> synchronousQueue = new SynchronousQueue<>();

        new Thread(()->{
            System.out.println(Thread.currentThread().getName() + " put 1");
            try {
                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() + synchronousQueue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName() + synchronousQueue.take());
                TimeUnit.SECONDS.sleep(3);
                System.out.println(Thread.currentThread().getName() + synchronousQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "T2").start();

    }

}

9. 线程池

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

池化技术

程序的运行，本质：占用系统资源。

优化资源的使用=>池化技术

线程池、连接池、内存池、对象池等。创建、销毁十分浪费资源。

池化技术：事先准备好一些资源，需要用就在池中取出，用完归还给池。

默认大小、最大值。

线程池的好处

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

线程复用，控制最大并发数，管理线程。

9.1. 三大方法

package com.gs.pool;

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

/**
 * Executors 工具类 三大方法
 *
 * 使用线程池之后要使用线程池来创建线程
 * @author admin
 * @date 2021/8/26 6:45 下午
 */
public class Demo01 {
    public static void main(String[] args) {
        // 单个线程
//        ExecutorService threadPool = Executors.newSingleThreadExecutor();

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

        try {
            for (int i = 0; i < 100; i++) {
                threadPool.execute(()->{
                    System.out.println(Thread.currentThread().getName() + " ok");
                });
            }
        } finally {
            // 线程池用完需要关闭
            threadPool.shutdown();
        }

    }
}

9.2. 七大参数

源码分析

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, // 约21亿 OOM
                                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;
}

keepAliveTime:

手动创建一个线程池

 package com.gs.pool;

 import java.util.concurrent.*;

 /**
  * Executors 工具类 七大参数
  *
  * @author admin
  * @date 2021/8/26 6:45 下午
  */
 public class Demo02 {
     public static void main(String[] args) {
         // 自定义线程池
         ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
                 2,
                 5,
                 3,
                 TimeUnit.SECONDS,
                 new LinkedBlockingQueue<>(3),   // 最大承载=队列数+最大线程值
                 Executors.defaultThreadFactory(),
                 new ThreadPoolExecutor.AbortPolicy()
                 );
         // 超出最大承载将报异常 RejectedExecutionException

         try {
             for (int i = 0; i < 100; i++) {
                 threadPool.execute(()->{
                     System.out.println(Thread.currentThread().getName() + " ok");
                 });
             }
         } finally {
             // 线程池用完需要关闭
             threadPool.shutdown();
         }

     }
 }

9.3. 四种拒绝策略

new ThreadPoolExecutor.AbortPolicy() // 如果队列满了，但是还有线程进来，不处理这个线程并且抛出异常
new ThreadPoolExecutor.CallerRunsPolicy() // 哪里来的去哪里
new ThreadPoolExecutor.DiscardPolicy() // 队列满了不会抛出异常，但超出的任务会直接丢弃
new ThreadPoolExecutor.DiscardOldestPolicy() // 队列满了，尝试去和最早的竞争。竞争失败就丢弃，也不会抛出异常

9.4. 小结

池的最大大小该怎么定义？（调优）

CPU密集型

几核CPU就定义为多少条，可以保证CPU效率最高。

获取CPU的核数：Runtime.getRuntime().availableProcessors()
IO密集型

判断程序中十分耗IO的线程有多少个，大于这个数（一般两倍）。

10. Fork/Join

10.1. 什么是Fork/Join？

Fork/Join在JDK1.7开始引入，并行执行任务。提高效率。适合大数据量。

大数据：Map Reduce（把大任务拆分为小任务）

10.2. Fork/Join特点

10.3. 工作窃取

即一个线程完成工作时，另一个线程没有完成工作，会将另一个线程的部分工作交给已经完成的线程完成。可以提高效率。

可以进行工作窃取的原因：这里面维护的都是双端队列。

10.4. Fork/Join的操作

ForkJoinPool

ForkJoinTask

RecursiveTask

package com.gs.forkjoin;

import java.util.concurrent.RecursiveTask;

/**
 * 求和计算任务
 *
 * Fork/Join
 * Stream并行流
 *
 * 如何使用fork/join
 * 1. forkjoinpool 通过它来执行
 * 2. 计算任务 forkjoinPool.execute(ForkJoinTask<?> task)
 * 3. 计算类要继承ForkJoinTask的实现类
 *
 * @author admin
 * @date 2021/8/26 11:59 下午
 */
public class ForkJoinDemo extends RecursiveTask<Long> {

    private Long start;
    private Long end;

    // 临界值
    private Long temp = 10_000L;

    public ForkJoinDemo(Long start, Long end) {
        this.start = start;
        this.end = end;
    }


    // 计算方法
    @Override
    protected Long compute() {
        if (end - start < temp) {
            // 单线程计算
            Long sum = 0L;
            for (Long i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        } else {
            // forkjoin

            // 中间值
            long middle = (start + end) / 2;

            // 拆分任务，把任务压入线程队列
            ForkJoinDemo task1 = new ForkJoinDemo(start, middle);
            ForkJoinDemo task2 = new ForkJoinDemo(middle + 1, end);
            task1.fork();
            task2.fork();

            // 线程相加并返回
            return task1.join() + task2.join();

        }
    }
}

package com.gs.forkjoin;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.stream.LongStream;

/**
 * @author admin
 * @date 2021/8/27 3:47 下午
 */
public class Test {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        test1();    // sum = 500000000500000000 时间：13249
//        test2();    // sum = 500000000500000000 时间：10952
        test3();    // sum = 500000000500000000 时间：1621

    }

    // 普通程序员
    public static void test1() {
        Long sum = 0L;
        long start = System.currentTimeMillis();
        for (int i = 0; i <= 1_000_000_000; i++) {
            sum += i;
        }
        long end = System.currentTimeMillis();
        System.out.println("sum = " + sum + " 时间：" + (end - start));
    }

    // 会使用forkjoin
    public static void test2() throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinDemo task = new ForkJoinDemo(0L, 1_000_000_000L);

        // 执行任务，没有返回结果
//        forkJoinPool.executoe(task);
        // 提交任务，有返回结果
        ForkJoinTask<Long> submit = forkJoinPool.submit(task);
        Long sum = submit.get();
        long end = System.currentTimeMillis();
        System.out.println("sum = " + sum + " 时间：" + (end - start));
    }

    // 使用Stream并行流
    public static void test3() {
        long start = System.currentTimeMillis();

        long sum = LongStream.rangeClosed(0L, 1_000_000_000L).parallel()0, Long::sum);

        long end = System.currentTimeMillis();
        System.out.println("sum = " + sum + " 时间：" + (end - start));
    }
}

11. 异步回调

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

package com.gs.future;


import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * 异步调用 CompletableFuture
 * <p>
 * 异步执行
 * 成功回调
 * 失败回调
 *
 * @author admin
 * @date 2021/8/27 7:56 下午
 */
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=>Void");
//        });
//
//        System.out.println(1111);
//
//        // 阻塞获取执行结果
//        completableFuture.get();

        // 有返回值的异步回调 supplyAsync
        // ajax 成功和失败的回调
        // 失败返回的是错误信息
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "completableFuture=>Integer");
            int i = 10 / 0; // 制造错误
            return 1024;
        });

        System.out.println(completableFuture.whenComplete((t, u) -> {
            System.out.println(t);  // 正常的返回结果
            System.out.println(u);  // 错误信息 java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
        }).exceptionally((E) -> {
            E.getMessage();
            return -1;  // 可以获取到错误的返回结果
        }).get());


        /**
         * success Code 200
         * error Code 404
         */
    }
}

12. Volatile

Volatile是Java虚拟机提供的轻量级的同步机制。

12.1. 特性

保证可见性。
不保证原子性。
禁止指令重排。

12.2. 特性验证

保证可见性验证

```java package com.gs.tvolatile;

import java.util.concurrent.TimeUnit;

/**

@author admin

@date 2021/8/28 12:24 下午 */ public class JMMDemo { private static int num = 0; // main线程 public static void main(String[] args) throws InterruptedException {

 new Thread(() -> {  // 线程1对主内存的变化不知道的
     while (num == 0) {

     }
 }).start();

 TimeUnit.SECONDS.sleep(1);

 num = 1;
 System.out.println(num);

 /*
 问题：main线程返回1，但是线程1一直在运行
 原因：线程1将num变量read和load到工作内存后，不会对main线程的原先变量进行监听
 需要线程1知道主内存中的值发生了变化
  */

} }


![image-20210828122948627](JUC.assets/image-20210828122948627.png)

```java
package com.gs.tvolatile;

import java.util.concurrent.TimeUnit;

/**
 * @author admin
 * @date 2021/8/28 12:24 下午
 */
public class JMMDemo {
    // 不加 volatile程序就会死循环
    // 加volatile可以保证可见性
    private volatile static int num = 0;
    // main线程
    public static void main(String[] args) throws InterruptedException {

        new Thread(() -> {  // 线程1
            while (num == 0) {

            }
        }).start();

        TimeUnit.SECONDS.sleep(1);

        num = 1;
        System.out.println(num);

    }
}

不保证原子性验证

原子性：

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

普通情况下

package com.gs.tvolatile;

/**
 * 普通情况下不保证原子性
 * @author admin
 * @date 2021/8/28 12:35 下午
 */
public class VDemo02 {

    private static int num = 0;

    public static void add() {
        num++;
    }

    public static void main(String[] args) {

        // 理论上结果应该为20_000
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    add();
                }
            }).start();
        }

        while (Thread.activeCount() > 2) {  // main gc
            Thread.yield();
        }

        System.out.println(Thread.currentThread().getName() + " " + num);   // main 19738

    }
}

加synchronized关键字情况下

package com.gs.tvolatile;

/**
 * synchronized关键字不保证原子性
 * @author admin
 * @date 2021/8/28 12:35 下午
 */
public class VDemo02 {

    private static int num = 0;

    public synchronized static void add() {
        num++;
    }

    public static void main(String[] args) {

        // 理论上结果应该为20_000
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    add();
                }
            }).start();
        }

        while (Thread.activeCount() > 2) {  // main gc
            Thread.yield();
        }

        System.out.println(Thread.currentThread().getName() + " " + num);   // main 20000

    }
}

加volatile关键字情况下

package com.gs.tvolatile;

/**
 * volatile关键字不保证原子性
 * @author admin
 * @date 2021/8/28 12:35 下午
 */
public class VDemo02 {

    private volatile static int num = 0;

    public static void add() {
        num++;
    }

    public static void main(String[] args) {

        // 理论上结果应该为20_000
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    add();
                }
            }).start();
        }

        while (Thread.activeCount() > 2) {  // main gc
            Thread.yield();
        }

        System.out.println(Thread.currentThread().getName() + " " + num);   // main 19657

    }
}

如果不加lock和synchronized，怎么保证原子性？

JUC包下的atomic包下，使用原子类解决原子性问题。

package com.gs.tvolatile;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * volatile更改为保证原子性
 * @author admin
 * @date 2021/8/28 12:35 下午
 */
public class VDemo02 {

    private volatile static AtomicInteger num = new AtomicInteger();

    public static void add() {
        num.getAndIncrement();  // AtomicInteger + 1 方法 CAS
    }

    public static void main(String[] args) {

        // 理论上结果应该为20_000
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                for (int j = 0; j < 1000; j++) {
                    add();
                }
            }).start();
        }

        while (Thread.activeCount() > 2) {  // main gc
            Thread.yield();
        }

        System.out.println(Thread.currentThread().getName() + " " + num);   // main 20000

    }
}

这些类的底层用的大部分是native方法，和操作系统挂钩。在内存中修改值。

Unsafe类是一个很特殊的存在。

public final int getAndIncrement() {
  return unsafe.getAndAddInt(this, valueOffset, 1);
}

指令重排

什么是指令重排？

编写的程序，计算机并不是按照编写的那样执行。

在不影响结果的前提下，对某些代码优先执行，用于提高效率。遵从JMM的happens-before原则。

源代码-->编译器优化的重排-->指令并行也可能会重排-->内存系统也会重排-->执行

加上volatile会在编译后的代码块前后加上内存屏障，禁止指令重排。

内存屏障是CPU指令，作用：

保证特定的操作的执行顺序。
可以保证某些变量的内存可见性（利用这些特性volatile实现了可见性）。

volatile的内存屏障在单例模式是用的最多。（DCL懒汉式）

13. CAS

什么是CAS

CAS是CPU的并发原语。

比较当前工作内存中的值和主内存中的值，如果这个值是期望的，则执行操作，否则一直循环。

缺点：

循环会耗时。
一次性只能保证一个共享变量的原子性。
存在ABA问题。

Unsafe类

package com.gs.cas;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * CAS
 * @author admin
 * @date 2021/8/28 4:46 下午
 */
public class CASDemo {


    public static void main(String[] args) {

        AtomicInteger atomicInteger = new AtomicInteger(2021);

        // 比较并交换
        // public final boolean compareAndSet(int expect, int update)
        // 如果和期望的值相同，就更新，否则就不更新
        System.out.println(atomicInteger.compareAndSet(2021, 2022));    // 2022
        System.out.println(atomicInteger.get());    // 2022

    }
}

14. ABA问题（狸猫换太子）

package com.gs.cas;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * CAS
 *
 * @author admin
 * @date 2021/8/28 4:46 下午
 */
public class CASDemo {


    public static void main(String[] args) {

        AtomicInteger atomicInteger = new AtomicInteger(11);

        // B
        System.out.println(atomicInteger.compareAndSet(11, 22));
        System.out.println(atomicInteger.get());

        System.out.println(atomicInteger.compareAndSet(22, 11));
        System.out.println(atomicInteger.get());

        //A

        System.out.println(atomicInteger.compareAndSet(11, 66));
        System.out.println(atomicInteger.get());

      // 会造成类似于SQL中的脏读问题

    }
}

CAS解决方法：使用原子引用。

15. 原子引用AtomicReference

带版本号的原子操作

package com.gs.cas;

import com.sun.codemodel.internal.fmt.JStaticFile;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicStampedReference;

/**
 * CAS
 *
 * @author admin
 * @date 2021/8/28 4:46 下午
 */
public class CASDemo {

    // Integer的坑：对象缓存机制 -128～127 超过会分配新内存空间
    // AtomicStampedReference如果泛型是包装类，注意对象的引用问题


    // 初始值，版本戳
    // 正常的业务操作，比较的是一个个对象
    static AtomicStampedReference<Integer> atomicStampedReference = new AtomicStampedReference<Integer>(11, 1);

    public static void main(String[] args) throws InterruptedException {

//        AtomicInteger atomicInteger = new AtomicInteger(2021);


        new Thread(() -> {
            int stamp = atomicStampedReference.getStamp();   // 获取版本戳
            System.out.println("A1 => " + stamp);


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

            System.out.println(atomicStampedReference.compareAndSet(11, 22, atomicStampedReference.getStamp(), atomicStampedReference.getStamp() + 1));

            stamp = atomicStampedReference.getStamp();   // 获取版本戳
            System.out.println("A2 => " + stamp);

            System.out.println(atomicStampedReference.compareAndSet(22, 11, atomicStampedReference.getStamp(), atomicStampedReference.getStamp() + 1));

            stamp = atomicStampedReference.getStamp();   // 获取版本戳
            System.out.println("A3 => " + stamp);
        }, "A").start();


        // 乐观锁的原理相同
        new Thread(() -> {
            int stamp = atomicStampedReference.getStamp();   // 获取版本戳
            System.out.println("B1 => " + stamp);

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

            System.out.println(atomicStampedReference.compareAndSet(11, 66, stamp, stamp + 1));

            stamp = atomicStampedReference.getStamp();   // 获取版本戳
            System.out.println("B2 => " + stamp);

        }, "B").start();

        /*
        A1 => 1
        B1 => 1
        true
        A2 => 2
        true
        A3 => 3
        false
        B2 => 3
         */
    }
}

16. 各种锁的理解

16.1. 公平锁和非公平锁

公平锁: 非常公平,先来后到,不允许插队

非公平锁: 非常不公平, 允许插队

public ReentrantLock() {
  sync = new NonfairSync(); //无参默认非公平锁
}
public ReentrantLock(boolean fair) {
  sync = fair ? new FairSync() : new NonfairSync();//传参为true为公平锁
}

16.2. 可重入锁

recursive 递归的

可重入锁（递归锁）

一个锁

package com.gs.lock;

/**
 * synchronized
 * @author admin
 * @date 2021/8/29 11:48 上午
 */
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");
    }
}

/*
Asms
Acall
Bsms
Bcall
 */

两个锁

package com.gs.lock;

import java.util.concurrent.locks.ReentrantLock;

/**
 * lock
 * @author admin
 * @date 2021/8/29 11:48 上午
 */
public class Demo02 {
    public static void main(String[] args) {
        Phone2 phone = new Phone2();
        new Thread(()->{
            phone.sms();
        },"A").start();

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

class Phone2 {
    ReentrantLock lock = new ReentrantLock();
    public void sms() {
        lock.lock();
        try {
            System.out.println(Thread.currentThread().getName() + "sms");
            call(); // 这里也有锁
        } finally {
            lock.unlock();
        }
    }

    public void call() {
        lock.lock();
        try {
            System.out.println(Thread.currentThread().getName() + "call");
        } finally {
            lock.unlock();
        }
    }
}
/*
Asms
Acall
Bsms
Bcall
 */

锁必须配对，否则会死锁。

16.3. 自旋锁

spinlock

自定义锁测试

package com.gs.lock;

import java.util.concurrent.atomic.AtomicReference;

/**
 * 自旋锁
 * @author admin
 * @date 2021/8/29 12:02 下午
 */
public class SpinLockDemo {

    // int 0
    // Thread null
    AtomicReference<Thread> atomicReference = new AtomicReference<>();

    // 加锁
    public void myLock() {
        Thread thread = Thread.currentThread();
        System.out.println(thread.getName() + "==> myLock");
        while (!atomicReference.compareAndSet(null, thread)) {

        }
    }

    // 解锁
    public void myUnlock() {
        Thread thread = Thread.currentThread();
        System.out.println(thread.getName() + "==> myUnlock");
        atomicReference.compareAndSet(thread, null);
    }

}

package com.gs.lock;

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

/**
 * @author admin
 * @date 2021/8/29 2:02 下午
 */
public class TestSpinLock {
    public static void main(String[] args) throws InterruptedException {
//        ReentrantLock reentrantLock = new ReentrantLock();
//        reentrantLock.lock();
//        try {
//
//        } finally {
//            reentrantLock.unlock();
//        }

        // 底层使用的自旋锁
        SpinLockDemo lock = new SpinLockDemo();

        new Thread(()->{
            lock.myLock();
            try {
                TimeUnit.SECONDS.sleep(3);

            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.myUnlock();
            }
        }, "T1").start();


        TimeUnit.SECONDS.sleep(1);

        new Thread(()->{
            lock.myLock();
            try {
                TimeUnit.SECONDS.sleep(1);

            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.myUnlock();
            }
        }, "T2").start();


    }
}

T1==> myLock
T2==> myLock
T1==> myUnlock
T2==> myUnlock

16.4. 死锁

死锁是什么？

死锁测试，怎么排除死锁？

package com.gs.lock;

import lombok.SneakyThrows;

import java.util.concurrent.TimeUnit;

/**
 * @author admin
 * @date 2021/8/29 2:25 下午
 */
public class DeadLockDemo {

    public static void main(String[] args) {
        String lockA = "lockA";
        String lockB = "lockB";

        new Thread(new MyThread(lockA, lockB), "T1").start();
        new Thread(new MyThread(lockB, lockA), "T2").start();
    }
}

class MyThread implements Runnable {

    private String lockA;
    private String lockB;

    public MyThread(String lockA, String lockB) {
        this.lockA = lockA;
        this.lockB = lockB;
    }

    @SneakyThrows
    @Override
    public void run() {
        synchronized (lockA) {
            System.out.println(Thread.currentThread().getName() + lockA);
            TimeUnit.SECONDS.sleep(2);
            synchronized (lockB) {
                System.out.println(Thread.currentThread().getName() + lockB);
            }
        }
    }
}

解决问题

使用jps -l定位进程号。

在死锁状态下，暂不要终止程序，使用终端jps -l命令可定位运行中的进程。

使用jstack 进程号查看进程信息，找到死锁问题。

JUC