Java基础多线程下篇

本篇本文目录:

- - 一.线程死锁
  - - 1.重入锁
    - 2.释放锁
    - 3.死锁
    - 4.wait和notify
  - 二.锁
  - - 1.ReentrantLock
    - 2.Condition
    - 3.ReadWriteLock
    - 4.StampedLock
    - 5.Semaphore
    - 6.线程并发安全
    - - (1) concurrent
      - (2) atomic
  - 三.线程池
  - - 1.通过线程池分配线程
    - 2.FixedThreadPool
    - 3.CachedThreadPool
    - 4.ScheduledThreadPool
  - 四.Callable接口
  - - 1.Callable接口
    - 2.Future
    - 4.ForkJoin
  - 五.ThreadLocal

一.线程死锁

1.重入锁

JVM允许同一个线程重复获取同一个锁，这种能被同一个线程反复获取的锁，就叫做可重入锁,Java的线程锁就是可重入锁，所以，获取锁的时候，不但要判断是否是第一次获取，还要记录这是第几次获取。每获取一次锁，记录+1，每退出synchronized块，记录-1，减到0的时候，才会真正释放锁。

例子:

当执行add(int n)方法,且n<0时会执行dec(-n)方法,由于在执行add(int n)方法时我们获取到了this这个对象的锁,当我们调用dec(-n)方法时我们任然可以再次获取this这个对象的锁,这就是重入锁,并且此时记录数为2。

public class Counter {
    private int count = 0;

    public synchronized void add(int n) {
        if (n < 0) {
            dec(-n);
        } else {
            count += n;
        }
    }

    public synchronized void dec(int n) {
        count += n;
    }
}

2.释放锁

当,当前线程中的记录数为0时就会释放锁。

下面操作会释放锁:

在这里插入图片描述
下面操作不会释放锁:

在这里插入图片描述

3.死锁

多个线程都占用了对方的锁资源,但不肯相让，导致了死锁,在编程是一定要避免死锁的发生.

应用案例:
在这里插入图片描述
实例代码如下:

package com.zm.synchronization;

import org.junit.Test;

public class DeadLockDemo extends Thread {
    private static Object playPhone = new Object();
    private static Object doHomerWork = new Object();
    private  boolean tage;

    public DeadLockDemo(boolean tage) {
        this.tage = tage;
    }
    @Override
    public void run() {
        if (tage) {
            // 先玩手机再做作业
            synchronized (playPhone) {
                System.out.println(Thread.currentThread().getName() + ":我在玩手机不要来管喔!");
                synchronized (doHomerWork) {
                    System.out.println(Thread.currentThread().getName() + ":妈妈,我手机玩好啦,开始做作业啦");
                }
            }

        } else {
            synchronized (doHomerWork) {
                System.out.println(Thread.currentThread().getName() +":儿子，快去做作业,做完再玩!");
                synchronized (playPhone) {
                    System.out.println(Thread.currentThread().getName() +":儿子真乖,去玩会手机吧");
                }
            }
        }
    }

    public static void main(String[] args) {
        DeadLockDemo mom = new DeadLockDemo(false);
        mom.setName("妈妈");
        DeadLockDemo son = new DeadLockDemo(true);
        son.setName("儿子");
        mom.start();
        son.start();
    }
}

运行效果( 造成了死锁 ):
在这里插入图片描述

备注:解决死锁的方法就是保证锁的顺序要一致。

4.wait和notify

多线程可以通过synchronized解决多线程竞争的问题,但是无法解决线程协调的问题,这时候可以使用wait和notify用于多线程协调运行：

在synchronized内部可以调用wait()使线程进入等待状态；
必须在已获得的锁对象上调用wait()方法；
在synchronized内部可以调用notify()或notifyAll()唤醒其他等待线程,其中notify()是随机唤醒一个,有可能某个线程一直无法唤醒,所以使用notifyAll()更安全一些；
必须在已获得的锁对象上调用notify()或notifyAll()方法；
已唤醒的线程还需要重新获得锁后才能继续执行。

例子:

下面的代码,大致就是先创建5个线程去调用getTask(),该方法的作用就是去删除队列里的数据 (该线程内部采用while(true)的死循环,当触发中断异常的时候才会结束线程),然后创建1个线程去向队列添加数据,共添加10条数据,添加完后结束该线程。

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.Queue;

public class ThreadTest {
    public static void main(String[] args) throws InterruptedException {
        // 创建一个TaskQueue的队列
        TaskQueue q = new TaskQueue();
        // 创建一个线程集合
        ArrayList<Thread> ts = new ArrayList<>();


        // 循环创建线程
        for (int i=0; i<5; i++) {
            // 创建执行getTask()方法的线程
            Thread t = new Thread(() -> {
                // 执行task:
                while (true) {
                    try {
                        String s = q.getTask();// 创建的线程去执行队列里的getTask()方法
                        System.out.println("线程:"+Thread.currentThread().getName()+"execute task: " + s);
                    } catch (InterruptedException e) {
                        System.out.println("退出getTask()线程!");
                        return; // 当外部嗲用中断方法时,退出执行
                    }
                }
            });
            t.start();// 启动线程
            ts.add(t);// 将该线程加入到集合中
        }

        // 创建执行addTask()的线程
        Thread add = new Thread(() -> {
            for (int i=0; i<10; i++) {
                // 放入task:
                String s = "t-" + Math.random();
                System.out.println("add task: " + s);
                q.addTask(s);// 将数据加入到队列中去
                try { Thread.sleep(100); } catch(InterruptedException e) {}
            }
        });
        add.start();// 启动线程
        add.join();// 当前线程插队

  		Thread.sleep(100);// 主线程休眠100ms
        for (Thread t : ts) {// 调用线程的中断方法,当前正在执行sleep/wait的线程会触发中断异常
            t.interrupt();
        }


    }
}

class TaskQueue {
   Queue<String> queue = new LinkedList<>();

    public synchronized void addTask(String s) {
        this.queue.add(s);
    }

    public synchronized String getTask() throws InterruptedException {
        if (!queue.isEmpty()) {
            return queue.remove();
        }
        return "当前队列没有任何参数,删除失败!";
    }
}

运行效果:

程序会一直循环的去执行 getTask()方法,虽然在主线程中调用了 t.interrupt();方法,但是并不会引发中断异常,因为需要线程正在执行sleep/wait时才会触发,所以这5个执行 getTask()的线程会一致执行下去。

在这里插入图片描述

当队列没有数据时,5个线程会一致执行,并且是无意义的执行,这样非常消耗CPU资源,这时我们可以通过wait方法去暂停该线程,直到队列里面有数据后才去执行(通过notifyAll()唤醒所有线程),如下:

在这里插入图片描述

运行效果:

程序执行后不在是死循环了,但是出现了相应的异常信息!

在这里插入图片描述

上面出现的异常信息是因为当执行addTask()方法时会让所有线程都唤醒,假设5个线程唤醒,队列里的数据只有3个时,当第四个线程从wait返回的时候,去执行remove()方法时就会抛出异常,因为队列里没有数据,if判断并没有判断到,因为用if判断只能判断一次，而且这一次在我们执行wait之前就已经执行,所以当我们返回的时候没有再一次判断当前队列中的数据是否为空,而直接去执行remove()方法,解决办法非常简单将if换成while即可,如下:

在这里插入图片描述
再次运行:

在这里插入图片描述

二.锁

1.ReentrantLock

从Java 5开始，引入了一个高级的处理并发的java.util.concurrent包，它提供了大量更高级的并发功能，能大大简化多线程程序的编写。我们知道Java语言直接提供了synchronized关键字用于加锁，但这种锁一是很重，二是获取时必须一直等待，没有额外的尝试机制。java.util.concurrent.locks包提供的ReentrantLock用于替代synchronized加锁。

传统的synchronized代码(前面售票的例子)：

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

public class SynChronZedTest {
    public static void main(String[] args) {
        RunnableNew runnableNew = new RunnableNew();
        Thread thread = new Thread(runnableNew);
        thread.setName("售票窗口1");
        thread.start();

        Thread thread2 = new Thread(runnableNew);
        thread2.setName("售票窗口2");
        thread2.start();

        Thread thread3 = new Thread(runnableNew);
        thread3.setName("售票窗口3");
        thread3.start();

        Thread thread4 = new Thread(runnableNew);
        thread4.setName("售票窗口4");
        thread4.start();
    }
}

class RunnableNew implements Runnable {
    private static Integer tick = 50;
    @Override
    public void run() {
        while (true) {
            synchronized (this) {
          
                if (tick <= 0) {
                    System.out.println("售票结束");
                    break;
                }

                try {
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                // 进行出售
                System.out.println(Thread.currentThread().getName() + "售出,还剩票数" + (--tick));
             
        }
        }
    }
}

如果用ReentrantLock替代，可以把代码改造为：

在这里插入图片描述

因为synchronized是Java语言层面提供的语法，所以我们不需要考虑异常，而ReentrantLock是Java代码实现的锁，我们就必须先获取锁，然后在finally中正确释放锁。顾名思义，ReentrantLock是可重入锁，它和synchronized一样，一个线程可以多次获取同一个锁。和synchronized不同的是，ReentrantLock可以尝试获取锁：

在这里插入图片描述

上述代码在尝试获取锁的时候，最多等待1秒。如果1秒后仍未获取到锁，tryLock()返回false，程序就可以做一些额外处理，而不是无限等待下去。所以，使用ReentrantLock比直接使用synchronized更安全，线程在tryLock()失败的时候不会导致死锁。

2.Condition

使用ReentrantLock比直接使用synchronized更安全，可以替代synchronized进行线程同步。但是，synchronized可以配合wait和notify实现线程在条件不满足时等待，条件满足时唤醒，用ReentrantLock我们怎么编写wait和notify的功能呢？答案是使用Condition对象来实现wait和notify的功能。

我们仍然以TaskQueue为例，把前面用synchronized实现的功能通过ReentrantLock和Condition来实现：

前面的实例:

在这里插入图片描述
修改后的:

在这里插入图片描述

可见，使用Condition时，引用的Condition对象必须从Lock实例的newCondition()返回，这样才能获得一个绑定了Lock实例的Condition实例。Condition提供的await()、signal()、signalAll()原理和synchronized锁对象的wait()、notify()、notifyAll()是一致的，并且其行为也是一样的：

await()会释放当前锁，进入等待状态；
signal()会唤醒某个等待线程；
signalAll()会唤醒所有等待线程；
唤醒线程从await()返回后需要重新获得锁。

此外，和tryLock()类似，await()可以在等待指定时间后，如果还没有被其他线程通过signal()或signalAll()唤醒，可以自己醒来：

if (condition.await(1, TimeUnit.SECOND)) {
    // 被其他线程唤醒
} else {
    // 指定时间内没有被其他线程唤醒
}

3.ReadWriteLock

在进行读操作的时候线程的同步并不重要,虽然保证写操作的同步性但是在一些使用读操作的场景下这样操作效率太低,我可以通过ReadWriteLock锁实现读取的时候可以有多个读操作,写的时候只有一个操作。

import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.IntStream;

public class Counter {
    private final ReadWriteLock rwlock = new ReentrantReadWriteLock();
    private final Lock rlock = rwlock.readLock();// 读锁
    private final Lock wlock = rwlock.writeLock();// 写锁
    private int[] counts = new int[2];

    public void inc(int index) {
        wlock.lock(); // 加写锁
        try {
            counts[index] += 1;
        } finally {
            wlock.unlock(); // 释放写锁
        }
    }

    public int[] get() {
        rlock.lock(); // 加读锁
        try {
            return Arrays.copyOf(counts, counts.length);
        } finally {
            rlock.unlock(); // 释放读锁
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Counter counter = new Counter();

        Thread writerThread = new Thread(()->{
            counter.inc(0);
            System.out.println("写入数据:(1,0)");
            counter.inc(1);
            System.out.println("写入数据:(1,1)");

        });
        writerThread.start();

        for (int i = 0; i < 1000; i++) {
            Thread readThread = new Thread(()->{
                int[] ints = counter.get();
                System.out.println("("+ints[0]+","+ints[1]+")");
            });
            readThread.start();
        }
    }
}

运行效果:

把读写操作分别用读锁和写锁来加锁，在读取时，多个线程可以同时获得读锁，这样就大大提高了并发读的执行效率。

在这里插入图片描述

4.StampedLock

如果我们深入分析ReadWriteLock，会发现它有个潜在的问题：如果有线程正在读，写线程需要等待读线程释放锁后才能获取写锁，即读的过程中不允许写，这是一种悲观的读锁。要进一步提升并发执行效率，Java 8引入了新的读写锁：StampedLock。 StampedLock和ReadWriteLock相比，改进之处在于：读的过程中也允许获取写锁后写入！这样一来，我们读的数据就可能不一致，所以，需要一点额外的代码来判断读的过程中是否有写入，这种读锁是一种乐观锁。乐观锁的意思就是乐观地估计读的过程中大概率不会有写入，因此被称为乐观锁。反过来，悲观锁则是读的过程中拒绝有写入，也就是写入必须等待。显然乐观锁的并发效率更高，但一旦有小概率的写入导致读取的数据不一致，需要能检测出来，再读一遍就行。

我们来看例子：

import java.util.concurrent.locks.StampedLock;

public class Point {
    private final StampedLock stampedLock = new StampedLock();// 乐观锁

    private double x;
    private double y;

    public void move(double deltaX, double deltaY) {
        long stamp = stampedLock.writeLock(); // 获取写锁
        try {
            x += deltaX;
            y += deltaY;
            System.out.println("写入数据:("+x+","+y+")");
        } finally {
            stampedLock.unlockWrite(stamp); // 释放写锁
        }
    }

    public void distanceFromOrigin() {
        long stamp = stampedLock.tryOptimisticRead(); // 获得一个乐观读锁
        // 注意下面两行代码不是原子操作
        // 假设x,y = (100,200)
        double currentX = x;
        // 此处已读取到x=100，但x,y可能被写线程修改为(300,400)
        double currentY = y;
        // 此处已读取到y，如果没有写入，读取是正确的(100,200)
        // 如果有写入，读取是错误的(100,400)
        if (!stampedLock.validate(stamp)) { // 检查乐观读锁后是否有其他写锁发生
            System.out.println("【验证失败】读取数据("+currentX+","+currentY+")");
            stamp = stampedLock.readLock(); // 获取一个悲观读锁
            try {
                currentX = x;
                currentY = y;
            } finally {
                stampedLock.unlockRead(stamp); // 释放悲观读锁
            }
        }else {
            System.out.println("【验证成功】读取数据("+currentX+","+currentY+")");
        }
    }

    public static void main(String[] args) {
        Point point = new Point();
        Thread writerThread= new Thread(()->{
            point.move(100,200);
            point.move(300,300);
            point.move(500,200);
        });
        writerThread.start();

        for (int i = 0; i < 1000; i++) {
            Thread readThread = new Thread(()->{
               point.distanceFromOrigin();
            });
            readThread.start();
        }
    }
}

运行效果:

和ReadWriteLock相比，写入的加锁是完全一样的，不同的是读取。注意到首先我们通过tryOptimisticRead()获取一个乐观读锁，并返回版本号。接着进行读取，读取完成后，我们通过validate()去验证版本号，如果在读取过程中没有写入，版本号不变，验证成功，我们就可以放心地继续后续操作。如果在读取过程中有写入，版本号会发生变化，验证将失败。在失败的时候，我们再通过获取悲观读锁再次读取。由于写入的概率不高，程序在绝大部分情况下可以通过乐观读锁获取数据，极少数情况下使用悲观读锁获取数据。可见，StampedLock把读锁细分为乐观读和悲观读，能进一步提升并发效率。但这也是有代价的：一是代码更加复杂，二是StampedLock是不可重入锁，不能在一个线程中反复获取同一个锁。StampedLock还提供了更复杂的将悲观读锁升级为写锁的功能，它主要使用在if-then-update的场景：即先读，如果读的数据满足条件，就返回，如果读的数据不满足条件，再尝试写。

在这里插入图片描述

5.Semaphore

使用锁的目的就是为了保证某一个资源在同一时刻只有一个线程能访问（ReentrantLock），或者只有一个线程能写入（ReadWriteLock）。还有一种受限资源，它需要保证同一时刻最多有N个线程能访问，比如同一时刻最多创建100个数据库连接，最多允许10个用户下载等。如果使用Lock数组来实现的话就太麻烦了，这个时候可以考虑使用Semaphore。

实例代码

首先创建一个Semaphore对象，参数为任意时刻仅允许访问的线程数量，然后调用semaphore.acquire(); 方法，调用acquire()可能会进入等待，直到满足条件为止。然后通过semaphore.release(); 进行释放。

package com.zm.semaphoretest;

import java.util.concurrent.*;

public class SemaphoreTest {
    // 任意时刻仅允许最多100个线程获取许可:
    final Semaphore semaphore = new Semaphore(100);
    static int count = 0;
    public static void main(String[] args) throws InterruptedException {
        for (int i = 1; i <= 1000; i++) {
            Thread thread = new Thread(()->{
                // 重写Runnable的run()
                try {
                    new SemaphoreTest().creatConnectDB();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            thread.start();
        }

        Thread.sleep(5000);
        System.out.println("创建了"+count+"个数据库连接:");


    }



    // 创建数据库连接
    public void creatConnectDB() throws InterruptedException {
        // 如果超过了许可数量,其他线程将在此等待:
        semaphore.acquire(); // 上锁
        try {
            // 创建数据库连接
            System.out.println(Thread.currentThread().getName()+"创建数据库连接");
            ++count;
        } catch (Exception e) {
            e.printStackTrace();
        }
        finally {
            semaphore.release();// 释放锁
        }
    }
}

运行效果:

在这里插入图片描述

可以通过 tryAcquire()来设置acquire()方法的等待时间(acquire()方法需要进行等待，知道满足条件后等待完毕)

    // 创建数据库连接
    public void creatConnectDB() throws InterruptedException {
        // 如果超过了许可数量,其他线程将在此等待:
        if (semaphore.tryAcquire(3, TimeUnit.SECONDS)) {
            try {
                // 创建数据库连接
                System.out.println(Thread.currentThread().getName()+"创建数据库连接");
                ++count;
            } catch (Exception e) {
                e.printStackTrace();
            }
            finally {
                semaphore.release();// 释放锁
            }
        }else{
            System.out.println(Thread.currentThread().getName()+"创建数据库失败，创建超时。");
        }
        
    }
}

6.线程并发安全

(1) concurrent

在并发环境下，尽量使用concurrent包中的集合对象，尽量避免自己去编写同步代码。

针对List、Map、Set、Deque等，java.util.concurrent包也提供了对应的并发集合类，如下:

在这里插入图片描述

因为所有的同步和加锁的逻辑都在集合内部实现，对外部调用者来说，只需要正常按接口引用，其他代码和原来的非线程安全代码完全一样。

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

使用 Collections.synchronizedMap(unsafeMap); 也能过实现线程安全，该方法实际上是用一个包装类包装了非线程安全的Map，然后对所有读写方法都用synchronized加锁，这样的方式虽然能保证安全但是，效率较低（悲观锁）。

Map unsafeMap = new HashMap();
Map threadSafeMap = Collections.synchronizedMap(unsafeMap);

(2) atomic

java.util.concurrent.atomic包下通过了一些子操作的封装类，通过这些封装类，也能够保证线程并发安全，这种方式，是无锁方式，效率上比synchronized加锁的效率高得多。（乐观锁）
商品超卖实例代码

package com.zm.atomictest;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * 原则操作
 */
public class AtomicTest {
    public static void main(String[] args) throws InterruptedException {
        RunnableNew runnableNew = new RunnableNew();
        Thread thread = new Thread(runnableNew);
        thread.setName("售票窗口1");
        thread.start();

        Thread thread2 = new Thread(runnableNew);
        thread2.setName("售票窗口2");
        thread2.start();

        Thread thread3 = new Thread(runnableNew);
        thread3.setName("售票窗口3");
        thread3.start();

        Thread thread4 = new Thread(runnableNew);
        thread4.setName("售票窗口4");
        thread4.start();

        Thread.sleep(5000);
        System.out.println("卖了:"+runnableNew.sum);

    }
}

class RunnableNew implements Runnable {
    AtomicInteger atom = new AtomicInteger(0);// 使用原子操作
    int tick=10,sum = 0;
    @Override
    public void run() {
        while (true) {
                if (tick <= 0) {
                    System.out.println("售票结束");
                    break;
                }else {
                    // 进行出售
                    ++sum;
                    System.out.println(Thread.currentThread().getName() + "售出,还剩票数" + (--tick));
                }

                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

        }


    }
}

运行效果,10张票卖出了12张票!

在这里插入图片描述

在上篇中我们通过同步锁解决了该问题

在这里插入图片描述

下面我们通过Atomic包提供的封装类来处理

首先通过atom.get()获取当前的版本值，通过加1的方式获取修改后的版本值，然后通过atom.compareAndSet(oldVersion, newVersion) 执行CAS操作，成功返回true，失败返回false,CAS操作就是通过将当前的oldVersion和内存中的oldVersion进行比较如果一致表示其他线程未对该变量进行修改，返回true,然后则将其替换为B。

在这里插入图片描述

三.线程池

1.通过线程池分配线程

在上文中三种创建线程的方式都是创建一个线程，然后去执行任务，任务执行完毕后关闭线程，如果下次使用的时候需要再次进行创建，这种方式资源消耗大,任务响应速度慢，可管理性差。我们可以通过池化技术来解决以上问题，就是通过线程池的方式去分配线程。简单地说，线程池内部维护了若干个线程，没有任务的时候，这些线程都处于等待状态。如果有新任务，就分配一个空闲线程执行。如果所有线程都处于忙碌状态，新任务要么放入队列等待，要么增加一个新线程进行处理。---廖雪峰官网使用线程池（https://www.liaoxuefeng.com/wiki/1252599548343744/1306581130018849）

Java标准库提供了ExecutorService接口表示线程池,该接口的几个常用实现类(以下实现类封装在Executors类中)如下:

FixedThreadPool：线程数固定的线程池；
CachedThreadPool：线程数根据任务动态调整的线程池；
SingleThreadExecutor：仅单线程执行的线程池。

2.FixedThreadPool

FixedThreadPool线程池表示创建固定大小的线程池

实例代码

package com.zm.createthread;

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

/**
 * 通过线程池的方式分配线程
 */
public class ExecutorServiceTest {
    public static void main(String[] args) {
        // 方式1
        ExecutorService executorService = Executors.newFixedThreadPool(5);// 线程池中有5个线程

        for (int i = 1; i <= 5; i++) {// 一共有5个任务
            executorService.submit(()->{
                try {
                    System.out.println(Thread.currentThread().getName()+":执行任务");
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        // 等任务执行完后，线程池关闭
        executorService.shutdown();



    }
}

运行效果:

在这里插入图片描述

当线程池的大小为3，而任务数为5时，剩余二个任务需要等待前面三个任务执行完后再进行执行:

在这里插入图片描述

注意submit()方法用于将任务提交到线程池中，其中接收的参数为Runnable和Callable接口。

在这里插入图片描述

shutdown()方法表示等任务执行完后，线程池再进行关闭，而shutdownNow()方法不需要等待全部任务执行完就关闭。

在这里插入图片描述

3.CachedThreadPool

CachedThreadPool线程池表示根据任务动态调整的线程池的大小

实例代码

package com.zm.createthread;

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

/**
 * 通过线程池的方式分配线程
 */
public class ExecutorServiceTest {
    public static void main(String[] args) {
        // 方式1
        //ExecutorService executorService = Executors.newFixedThreadPool(3);// 线程池中有5个线程
        // 方式2
        ExecutorService executorService = Executors.newCachedThreadPool();//根据任务数动态跳转任务数
        for (int i = 1; i <= 5; i++) {// 一共有5个任务
            executorService.submit(()->{
                try {
                    System.out.println(Thread.currentThread().getName()+":执行任务");
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        
        executorService.shutdown();  // 表示等任务执行完后，线程池在进行关闭
        //executorService.shutdownNow();// 表示立即关闭，不需要等待全部任务执行完


    }
}

运行效果

在这里插入图片描述

调整任务数量为10时，线程池的大小为10

在这里插入图片描述

创建指定动态范围的线程池

根据newCachedThreadPool()的源码，可知线程池通过ThreadPoolExecutor进行创建。

在这里插入图片描述

只需要修改第一个参数和第二个参数就可以实现在指定动态范围创建线程池

package com.zm.createthread;

import java.util.concurrent.*;

/**
 * 通过线程池的方式分配线程
 */
public class ExecutorServiceTest {
    public static void main(String[] args) {
        // 方式1
        //ExecutorService executorService = Executors.newFixedThreadPool(3);// 线程池中有5个线程
        // 方式2
       // ExecutorService executorService = Executors.newCachedThreadPool();//根据任务数动态跳转任务数
        int min = 0;
        int max = 10;
        ExecutorService executorService = new ThreadPoolExecutor(min, max,
                60L, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>());
        for (int i = 1; i <= 5; i++) {// 一共有5个任务
            executorService.submit(()->{
                try {
                    System.out.println(Thread.currentThread().getName()+":执行任务");
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        executorService.shutdown();  // 表示等任务执行完后，线程池在进行关闭
        //executorService.shutdownNow();// 表示立即关闭，不需要等待全部任务执行完


    }
}

在这里插入图片描述

4.ScheduledThreadPool

SingleThreadExecutor线程池表示仅单线程执行的线程池，意思是同一个任务需要反复执行(定时器)，就需要使用该线程池来实现。通过ScheduledThreadPool来实现单线程执行的线程池，其实ScheduledThreadPool就是在ExecutorService接口中多了几个方法，来实现定时执行任务。通过ScheduledThreadPool执行定时任务，是完全可以通过ScheduledThreadPool取代旧的Timer。

n秒/毫秒/分后执行一次性任务（只执行一次）:

参数corePoolSize – 要保留在池中的线程数，即使它们处于空闲状态

class ScheduledThreadPoolTest{
    public static void main(String[] args) {
        ScheduledExecutorService ses = Executors.newScheduledThreadPool(4);
        DateFormat dateFormat = new SimpleDateFormat("YYYY-MM-DD hh:mm:ss");
        System.out.println("当前时间:"+dateFormat.format(new Date()));
        // 1秒后执行一次性任务:
        ses.schedule(()->{
            System.out.println("当前时间:"+dateFormat.format(new Date()));
        }, 1, TimeUnit.SECONDS);
        ses.shutdown();// 当任务执行完毕后进行关闭
    }
}

在这里插入图片描述

n秒/毫秒/分后开始执行定时任务，每n秒/毫秒/分执行:

class ScheduledThreadPoolTest{
    public static void main(String[] args) {
        ScheduledExecutorService ses = Executors.newScheduledThreadPool(4);
        DateFormat dateFormat = new SimpleDateFormat("YYYY-MM-DD hh:mm:ss");
        System.out.println("当前时间:"+dateFormat.format(new Date()));
        // 2秒后开始执行定时任务，每3秒执行一次:
        ses.scheduleAtFixedRate(()->{
            System.out.println("当前时间:"+dateFormat.format(new Date()));
        }, 2, 3, TimeUnit.SECONDS);
    }
}

在这里插入图片描述

任务以固定的n秒/毫秒/分为间隔执行：

class ScheduledThreadPoolTest{
    public static void main(String[] args) {
        ScheduledExecutorService ses = Executors.newScheduledThreadPool(4);
        DateFormat dateFormat = new SimpleDateFormat("YYYY-MM-DD hh:mm:ss");
        System.out.println("当前时间:"+dateFormat.format(new Date()));
        // 2秒后开始执行定时任务，以3秒为间隔执行:
        ses.scheduleWithFixedDelay(()->{
            System.out.println("当前时间:"+dateFormat.format(new Date()));
        }, 2, 3, TimeUnit.SECONDS);
    }
}

在这里插入图片描述

FixedRate和FixedDelay的区别

FixedRate是指任务总是以固定时间间隔触发，不管任务执行多长时间：

在这里插入图片描述

而FixedDelay是指，上一次任务执行完毕后，等待固定的时间间隔，再执行下一次任务：

在这里插入图片描述

在FixedRate模式下，假设每秒触发，如果某次任务执行时间超过1秒，后续任务会不会并发执行？

class ScheduledThreadPoolTest{
    public static void main(String[] args) {
        ScheduledExecutorService ses = Executors.newScheduledThreadPool(4);
        DateFormat dateFormat = new SimpleDateFormat("YYYY-MM-DD hh:mm:ss");
        //2s后，每间隔一秒执行一次任务
        ses.scheduleAtFixedRate(()->{
            try {
                System.out.println("开始执行");
                Thread.sleep(2000);// 延时2s
                System.out.println("当前时间:"+dateFormat.format(new Date()));
                System.out.println("执行结束");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, 2, 1, TimeUnit.SECONDS);

    }
}

本来每1s执行一次任务，但是由于任务加了延时函数，所以需要2s才能够完成，根据下面的运行情况，可知后续任务并不会并行执行，而是会等上一次执行完毕后，再立即执行下一次。

在这里插入图片描述

如果任务抛出了异常，后续任务是否继续执行？

class ScheduledThreadPoolTest{
    public static void main(String[] args) throws InterruptedException {
        ScheduledExecutorService ses = Executors.newScheduledThreadPool(4);
        DateFormat dateFormat = new SimpleDateFormat("YYYY-MM-DD hh:mm:ss");
        ses.scheduleAtFixedRate(()->{
            System.out.println("开始执行");
            System.out.println("执行任务:1/0");
            int num1 = 1;
            int num2 = 0;
            int i = num1 / num2;
            System.out.println("开始结束");
        }, 2, 1, TimeUnit.SECONDS);

        Thread.sleep(5000);
        ses.shutdown();
    }
}

如果执行任务抛出了异常，后续任务不会继续执行。

在这里插入图片描述

四.Callable接口

1.Callable接口

在上文中submit()方法的参数有二种类型，Runnable和Callable接口，二个接口都是用来执行提交的任务，和Runnable接口不同的是Callable接口可以获取执行任务后的结果。

实例代码

package com.zm.createthread;

import java.util.concurrent.*;

public class CallableTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(5);// 线程池中有5个线程
        // 一个Future类型的实例代表一个未来能获取结果的对象
        Future<String> future = executorService.submit(new Task());// 通过Callable接口作为参数传入
        // 判断是否执行完毕
        while (true) {
            if (future.isDone()) {
                // 在调用get()时，如果异步任务已经完成，我们就直接获得结果。如果异步任务还没有完成，那么get()会阻塞，直到任务完成后才返回结果
                System.out.println("执行情况:"+future.get());
                executorService.shutdownNow();// 立即关闭线程池
                break;
            }
        }




    }
}
class Task implements Callable<String> {
    @Override
    public String call() {
        System.out.println("执行任务");
        return "成功";
    }
}

运行效果

在这里插入图片描述

如何获取执行任务返回的结果

executorService.submit(new Task())会返回一个 Future对象，该对象代表一个未来能获取结果的对象，通过Future对象的get()方法就可以获取到执行任务完成后返回的值(值的类型由Callable的返回类型决定，Callable<String> 表示返回一个String类的数据)，需要注意的是在调用get()时，如果异步任务已经完成，我们就直接获得结果。如果异步任务还没有完成，那么get()会阻塞，直到任务完成后才返回结果。所以会搭配Future对象的isDone()方法一起使用，isDone()表示是否执行完毕。

2.Future

Future接口定义了主要的5个接口方法，有RunnableFuture和SchedualFuture继承这个接口，以及CompleteFuture和ForkJoinTask实现这个接口。来源于: https://blog.csdn.net/u014209205/article/details/80598209

在这里插入图片描述

Future接口主要包括5个方法

方法	描述
get（）	可以当任务结束后返回一个结果，如果调用时，工作还没有结束，则会阻塞线程，直到任务执行完毕
get（long timeout,TimeUnit unit）	等待timeout的时间就会返回结果
cancel（boolean mayInterruptIfRunning）	可以用来停止一个任务，如果任务可以停止（通过mayInterruptIfRunning来进行判断），则可以返回true,如果任务已经完成或者已经停止，或者这个任务无法停止，则会返回false.
isDone（）	判断当前方法是否完成
isCancel（）	判断当前方法是否取消

CompletableFuture

从Java 8开始引入了CompletableFuture，它针对Future做了改进，可以传入回调对象，当异步任务完成或者发生异常时，自动调用回调对象的回调方法。

class CompletableFutureTest{
    public static void main(String[] args) throws InterruptedException {
        // 创建异步执行任务:
        CompletableFuture<Double> cf = CompletableFuture.supplyAsync(CompletableFutureTest::fetchPrice);
        // 如果执行成功:
        cf.thenAccept((result) -> {
            System.out.println("price: " + result);
        });
        // 如果执行异常:
        cf.exceptionally((e) -> {
            e.printStackTrace();
            return null;
        });

        // 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:
        Thread.sleep(200);
    }


    static Double fetchPrice() {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
        }
        if (Math.random() < 0.3) {
            throw new RuntimeException("fetch price failed!");
        }
        return 5 + Math.random() * 20;
    }
}

二种不同的执行效果:

在这里插入图片描述

CompletableFuture相较于Future还可以实现任务串行和并行。

例如，定义两个CompletableFuture，第一个CompletableFuture根据证券名称查询证券代码，第二个CompletableFuture根据证券代码查询证券价格，这两个CompletableFuture实现串行操作如下（ 串行化 ）：

实例代码:

class CompletableFutureTest2{
    public static void main(String[] args) throws Exception {
        // 第一个任务:
        CompletableFuture<String> cfQuery = CompletableFuture.supplyAsync(() -> {
            return queryCode("中国石油");
        });
        // cfQuery成功后继续执行下一个任务:
        CompletableFuture<Double> cfFetch = cfQuery.thenApplyAsync((code) -> {
            return fetchPrice(code);
        });
        // cfFetch成功后打印结果:
        cfFetch.thenAccept((result) -> {
            System.out.println("price: " + result);
        });
        // 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:
        Thread.sleep(2000);
    }

    static String queryCode(String name) {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
        }
        return "601857";
    }

    static Double fetchPrice(String code) {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
        }
        return 5 + Math.random() * 20;
    }
}

运行效果:

在这里插入图片描述

例如，我们考虑这样的场景：同时从新浪和网易查询证券代码，只要任意一个返回结果，就进行下一步查询价格，查询价格也同时从新浪和网易查询，只要任意一个返回结果，就完成操作（并行化）：

class CompletableFutureTest3{
    public static void main(String[] args) throws Exception {
        // 两个CompletableFuture执行异步查询:
        CompletableFuture<String> cfQueryFromSina = CompletableFuture.supplyAsync(() -> {
            return queryCode("中国石油", "https://finance.sina.com.cn/code/");
        });
        CompletableFuture<String> cfQueryFrom163 = CompletableFuture.supplyAsync(() -> {
            return queryCode("中国石油", "https://money.163.com/code/");
        });

        // 用anyOf合并为一个新的CompletableFuture:
        CompletableFuture<Object> cfQuery = CompletableFuture.anyOf(cfQueryFromSina, cfQueryFrom163);

        // 两个CompletableFuture执行异步查询:
        CompletableFuture<Double> cfFetchFromSina = cfQuery.thenApplyAsync((code) -> {
            return fetchPrice((String) code, "https://finance.sina.com.cn/price/");
        });
        CompletableFuture<Double> cfFetchFrom163 = cfQuery.thenApplyAsync((code) -> {
            return fetchPrice((String) code, "https://money.163.com/price/");
        });

        // 用anyOf合并为一个新的CompletableFuture:
        CompletableFuture<Object> cfFetch = CompletableFuture.anyOf(cfFetchFromSina, cfFetchFrom163);

        // 最终结果:
        cfFetch.thenAccept((result) -> {
            System.out.println("price: " + result);
        });
        // 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:
        Thread.sleep(200);
    }

    static String queryCode(String name, String url) {
        System.out.println("query code from " + url + "...");
        try {
            Thread.sleep((long) (Math.random() * 100));
        } catch (InterruptedException e) {
        }
        return "601857";
    }

    static Double fetchPrice(String code, String url) {
        System.out.println("query price from " + url + "...");
        try {
            Thread.sleep((long) (Math.random() * 100));
        } catch (InterruptedException e) {
        }
        return 5 + Math.random() * 20;
    }
}

运行效果:

在这里插入图片描述

汇总

thenAccept()处理正常结果；
exceptional()处理异常结果；
thenApplyAsync()用于串行化另一个CompletableFuture，当前一个CompletableFuture执行成功后才执行下一个CompletableFuture；
anyOf()和allOf()用于并行化多个CompletableFuture；anyOf()任意个CompletableFuture只要一个成功”，allOf()可以实现“所有CompletableFuture都必须成功。

4.ForkJoin

Java 7开始引入了一种新的Fork/Join线程池，它可以执行一种特殊的任务：把一个大任务拆成多个小任务并行执行。

实例代码,计算数组的和(数组大小2000)

package com.zm.forkJointest;

import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

public class ForkJoinTest {
    public static void main(String[] args) {
        // 创建2000个随机数组成的数组:
        long[] array = new long[2000];
        long expectedSum = 0;
        for (int i = 0; i < array.length; i++) {
            array[i] = random();
            expectedSum += array[i];
        }
        System.out.println("Expected sum: " + expectedSum+"消耗时间:");// 从0~1999进行累加


        // fork/join:
        ForkJoinTask<Long> task = new SumTask(array, 0, array.length);
        long startTime = System.currentTimeMillis();
        Long result = ForkJoinPool.commonPool().invoke(task);
        long endTime = System.currentTimeMillis();
        System.out.println("Fork/join sum: " + result + " in " + (endTime - startTime) + " ms.");
    }


    static Random random = new Random(0);
    static long random() {
        return random.nextInt(10000);
    }
}


class SumTask extends RecursiveTask<Long> {
    static final int THRESHOLD = 500;// 任务划分小任务的门槛
    long[] array;// 数组
    int start;// 开始位置
    int end;// 结束位置

    SumTask(long[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }

    @Override
    protected Long compute() {
        // 如果数组的大小小于THRESHOLD，就不需要划分为小任务，直接计算即可
        if (end - start <= THRESHOLD) {
            // 如果任务足够小,直接计算:
            long sum = 0;
            for (int i = start; i < end; i++) {
                sum += this.array[i];
                // 故意放慢计算速度:
                try {
                    Thread.sleep(1);
                } catch (InterruptedException e) {
                }
            }
            return sum;
        }

        // 任务太大,一分为二:
        int middle = (end + start) / 2;
        System.out.println(String.format("split %d~%d ==> %d~%d, %d~%d", start, end, start, middle, middle, end));
        SumTask subtask1 = new SumTask(this.array, start, middle);// 子任务1
        SumTask subtask2 = new SumTask(this.array, middle, end);// 子任务2
        invokeAll(subtask1, subtask2);// 并行运行二个任务
        Long subresult1 = subtask1.join();// 获取子任务1的运算结果
        Long subresult2 = subtask2.join(); // 获取子任务2的运算结果
        Long result = subresult1 + subresult2; // 将子任务1和子任务2累加
        System.out.println("result = " + subresult1 + " + " + subresult2 + " ==> " + result);
        return result;
    }
}

运行效果

在这里插入图片描述

五.ThreadLocal

同一个线程中，怎么将对象在各个方法中(上下文中)进行传递，首先会想到的就是通过将该参数作为方法的形参进行传递，但是一个方法可能包含多个方法，不可能为了传递一个参数让参数作为所有方法的形参吧，这样实在是太麻烦了。Java标准库提供了一个特殊的ThreadLocal，它可以在一个线程中传递同一个对象，所以我们可以通过ThreadLocal来实现同一个线程中对象的传递,ThreadLocal其实只是线程ThreadLocalMqp中的一个元素，key=当前线程本身,value=传递对象。相关博文:https://zhuanlan.zhihu.com/p/158033837

在这里插入图片描述

实例代码(模拟用户登入和登出的过程)

package com.zm.threadLocaltest;

import java.util.Scanner;

public class ThreadLocalTest {
    public static void main(String[] args) throws InterruptedException {
        // 进行登入
        LoginUser login = new LoginTask().login();
        System.out.println("【主线程】用户登入:" + login);
        // 主线程
        UserContext userContext = new UserContext(login);// 将当前用户对象存入到ThreadLocal
        System.out.println("【主线程】当前登入用户:" + UserContext.currentUser());


        // 其他线程
        System.out.println("其他线程");
        Thread thread = new Thread(()->{
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("【子线程】当前登入用户:" + UserContext.currentUser());
        });
        thread.start();

        Thread.sleep(5000);

        new LoginTask().logout(userContext);// 退出登入
        System.out.println("【主线程】当前登入用户:" + UserContext.currentUser());





    }
}


class LoginTask {
    // 进行登入
    public LoginUser login() {
        // 进行登入
        String userName = "张三";
        Integer passWord = 123456;
        String token = "aiaojcoajvcwpnkvanwknvklavnaklvnaklvnaklwwuduiwuwijf";
        // 进行登入
        LoginUser loginUser = new LoginUser(userName, passWord, token);
        return loginUser;
    }

    // 进行登出
    public void logout(UserContext currentUser) {
        System.out.println("用户退出登入");
        currentUser.close();// 移除
    }

}


/**
 * 通过实现AutoCloseable，就可以通过try (resource) {...}结构来实现自动关闭
 */
class UserContext implements AutoCloseable {

    static final ThreadLocal<LoginUser> ctx = new ThreadLocal<>();

    /**
     * 需要传递的对象
     * @param user 传递对象
     */
    public UserContext(LoginUser user) {
        ctx.set(user);
    }

    /**
     * 获取当前对象
     * @return 当前用户
     */
    public static LoginUser currentUser() {
        return ctx.get();
    }

    /**
     * 将当前对象进行移除
     */
    @Override
    public void close() {
        ctx.remove();
    }

}


/**
 * 当前登入用户的信息
 */
class LoginUser{
    private String userName;
    private Integer passWord;
    private String token;

    public LoginUser(String userName, Integer passWord, String token) {
        this.userName = userName;
        this.passWord = passWord;
        this.token = token;
    }

    public String getUserName() {
        return userName;
    }

    public void setUserName(String userName) {
        this.userName = userName;
    }

    public Integer getPassWord() {
        return passWord;
    }

    public void setPassWord(Integer passWord) {
        this.passWord = passWord;
    }

    public String getToken() {
        return token;
    }

    public void setToken(String token) {
        this.token = token;
    }


    @Override
    public String toString() {
        return "LoginUser{" +
                "userName='" + userName + '\'' +
                ", passWord=" + passWord +
                ", token='" + token + '\'' +
                '}';
    }
}