jdk中的简单并发，需要掌握

Posted 2021-02-02 youzhibing

前言

　　开心一刻

　　　　小时候有一次爸爸带我去偷村头别人家的梨子，我上树摘，爸爸在下面放风，正摘着主人来了，爸爸指着我破口大骂：臭小子，赶紧给我滚下来，敢偷吃别人家梨子，看我不打死你。主人家赶紧说：没事没事，小孩子淘气嘛，多摘点回家吃。我……这坑儿子的爹...

纯正的海豹突击队

　　路漫漫其修远兮，吾将上下而求索！

　　github：https://github.com/youzhibing

　　码云(gitee)：https://gitee.com/youzhibing

Runnable

　　如果是简单的实现一个线程，我们会通过实现Runnable接口或继承Thread类来完成。JDK1.0中就已经存在Runnable和Thread，Thread实现了Runnable接口。Runnable使用方式一般如下

public class RunnableTest {

    public static void main(String[] args) {
        // Java 8之前：
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("Before Java8, 我是子线程1");
            }
        }).start();

        //Java 8方式：
        new Thread( () -> {
            System.out.println("In Java8");
            System.out.println("我是子线程2");
        } ).start();

    }
}

View Code

　　一般我们的线程不是以匿名内部类的方式存在的，而是以如下方式存在

public class RunnableTest {

    public static void main(String[] args) throws InterruptedException {
        Runnable myRunnable = new MyRunnable();
        new Thread(myRunnable).start();
        Thread.sleep(1000);
        System.out.println("我是主线程");
    }
}
class MyRunnable implements Runnable {

    @Override
    public void run() {
        System.out.println("我是子线程1");
    }
}

View Code

　　当然线程的实现方式还有Thread类，Thread实现了Runnable接口，本质还是一样；无论是Runnable，还是Thread，实现的线程有一个很明显的缺点，就是没有返回值，执行完任务之后无法获取执行结果。

Callable

　　Callable接口是JDK1.5中引入的，和Runnable类似，都是用来实现多线程，不同的是，Callable能返回结果和抛出checked exception。源代码如下

@FunctionalInterface
public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

　　可以看到，Callable是一个泛型接口，call()函数返回的类型就是传递进来的泛型类型，也是返回的结果类型。那么怎么使用Callable呢？一般情况下是配合ExecutorService来使用的，而ExecutorService的创建又是用Executors来完成的。

线程池

　　Executors

　　　　也是JDK1.5新增内容，是创建ExecutorService、ScheduledExecutorService、ThreadFactory和Callable的工厂，并提供了一些有效的工具方法。有很多创建ExecutorService的方法

　　　　主要分为6类方法，每一类都两两重载，一个有ThreadFactory threadFactory参数，一个没有ThreadFactory threadFactory参数，也就是我们可以自定义ThreadFactory来定制Thread；若没有ThreadFactory参数，则使用默认的DefaultThreadFactory来构建Thread。6类方法如下

　　　　　　newCachedThreadPool(...)

　　　　　　　　创建一个可缓存的线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程；返回类型是：ThreadPoolExecutor。

　　　　　　newFixedThreadPool(...)

　　　　　　　　创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待；返回类型是：ThreadPoolExecutor。

　　　　　　newScheduledThreadPool(...)

　　　　　　　　创建一个定长线程池，支持定时及周期性任务执行；返回类型是：ScheduledThreadPoolExecutor。多数情况下可用来替代Timer类。

　　　　　　newSingleThreadExecutor(...)

　　　　　　　　创建一个单线程化的线程池，只有唯一的一个工作线程来执行任务，保证所有任务按照指定顺序执行；返回类型是：ThreadPoolExecutor的代理，我们可以认为就是ThreadPoolExecutor。

　　　　　　newSingleThreadScheduledExcutor(...)

　　　　　　　　创建一个单线程化的线程池，与newSingleThreadExecutor类似，但支持定时及周期性任务执行；返回类型是：ScheduledThreadPoolExecutor。

　　　　　　newWorkStealingPool(...)

　　　　　　　　创建持有足够线程的线程池来支持给定的并行级别，并通过使用多个队列，减少竞争；它需要穿一个并行级别的参数，如果不传，则被设定为默认的CPU数量。JDK1.8中新增，返回类型是：ForkJoinPool。ForkJoinFool通常配合ForkJoinTask的子类RecursiveAction或RecursiveTask使用。

　　　　常用的主要是以下3类：newCachedThreadPool,newFixedThreadPool,newScheduledThreadPool。至于newWorkStealingPool，我还没用过，不太好评论。

　　ExecutorService

　　　　ExecutorService是一个interface，继承了Executor，是Java中对线程池定义的一个接口，类图如下：

　　　　ExecutorService接口中常用方法如下

void execute(Runnable command);    // 从Executor继承而来，用来执行Runnale，没有返回值
<T> Future<T> submit(Callable<T> task);    // 执行Callable类型的task，并返回Future
<T> Future<T> submit(Runnable task, T result);    // 这种方式很少使用
Future<?> submit(Runnable task);    // 执行Runnable类型的task，并返回Future

　　　　当然还有invokeAll、invokeAny，感兴趣的可以去看下。关于Future，下面会讲到。

　　　　当我们使用完成ExecutorService之后应该关闭它，否则它里面的线程会一直处于运行状态，导致应用无法停止。关闭ExecutorService的方式有两种，其一是ExecutorService.shutdown()方法，在调用shutdown()方法之后，ExecutorService不会立即关闭，但是它不再接收新的任务，直到当前所有线程执行完成才会关闭，所有在shutdown()执行之前提交的任务都会被执行；其二是调用ExecutorService.shutdownNow()方法，它将跳过所有正在执行的任务和被提交还没有执行的任务，但是它并不对正在执行的任务做任何保证，有可能它们都会停止，也有可能执行完成。一般推荐的关闭方式是ExecutorService.shutdown()。

　　Future

　　　　对具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。相关类图如下

public interface Future<V> {

    /**
     * 取消任务，如果取消任务成功则返回true，如果取消任务失败则返回false
     * @param mayInterruptIfRunning 是否允许取消正在执行却没有执行完毕的任务，如果设置true，则表示可以取消正在执行过程中的任务。
     *         如果任务已经完成，则无论mayInterruptIfRunning为true还是false，此方法肯定返回false，即如果取消已经完成的任务会返回false。
     *         如果任务正在执行，若mayInterruptIfRunning设置为true，则返回true，若mayInterruptIfRunning设置为false，则返回false。
     *         如果任务还没有执行，则无论mayInterruptIfRunning为true还是false，肯定返回true。
     * @return
     */
    boolean cancel(boolean mayInterruptIfRunning);

    /**
     * 任务是否被取消成功，如果在任务正常完成前被取消成功，则返回true。
     * @return
     */
    boolean isCancelled();

    /**
     * 任务是否已经完成，若完成则返回true。
     * @return
     */
    boolean isDone();

    /**
     * 获取执行结果，这个方法会产生阻塞，会一直等到任务执行完毕才返回。
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    V get() throws InterruptedException, ExecutionException;

    /**
     * 获取执行结果，如果在指定时间内没获取到结果，就直接返回null
     * @param timeout
     * @param unit
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     * @throws TimeoutException
     */
    V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;
}

View Code

　　　　从如上代码可以看出Future提供了三种功能：

　　　　　　1、判断任务是否完成；2、中断任务；3、获取任务执行结果。

线程池使用示例

　　Runnable使用示例

　　　　示例一，定时周期的执行某个任务

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class RunnableTest {

    public static void main(String[] args) {
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(5);
        MyRunnable myRunnable = new MyRunnable();
        // 应用启动3秒开始执行myRunnable，之后每隔5秒执行一次; scheduleAtFixedRate是有返回值的，配合Runnable的话，我们不关注返回值
        executorService.scheduleAtFixedRate(myRunnable, 3, 5, TimeUnit.SECONDS);
        System.out.println("我是主线程...");
    }
}
class MyRunnable implements Runnable {

    @Override
    public void run() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("我是子线程1");
    }
}

View Code

　　　　示例二，单线程化的线程池执行某个任务，并显示的关闭线程池

public class RunnableTest {

    public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        MyRunnable myRunnable = new MyRunnable();

        // 为了可取消性而使用Future但又不提供可用的结果，则可以声明 Future<?> 形式类型、并返回null作为底层任务的结果
        Future<?> submit = executorService.submit(myRunnable);

        // 如果不shutdown，那它里面的线程会一直处于运行状态，应用不会停止
        executorService.shutdown();

        // 输出任务执行结果，由于Runnable没有返回值，所以get的是null
        System.out.println(submit.get(4, TimeUnit.SECONDS));
        System.out.println("我是主线程...");
    }
}
class MyRunnable implements Runnable {

    @Override
    public void run() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("我是子线程1");
    }
}

View Code

　　Callable使用示例

　　　　示例一，Callable + Future获取结果；采用缓存线程池执行任务

import java.util.concurrent.*;
import java.util.concurrent.Future;

public class CallableTest {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        Task task = new Task();
        Future<String> result = executorService.submit(task);
        executorService.shutdown();

        Thread.sleep(1000);
        System.out.println("我是主线程, 执行另外的业务...");

        System.out.println("task执行结果：" + result.get());

        System.out.println("任务全部执行完毕...");
    }
}
class Task implements Callable<String> {

    @Override
    public String call() throws Exception {
        System.out.println("子线程, 业务处理中...");
        Thread.sleep(2000);
        return "业务执行成功";
    }
}

View Code

　　　　示例二，Callable + FutureTask获取结果；采用定长线程池执行定时任务

import java.util.concurrent.*;

public class CallableTest {

    public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        Task task = new Task();
        FutureTask<String> futureTask = new FutureTask<>(task);
        executorService.submit(futureTask);
        executorService.shutdown();

        Thread.sleep(1000);
        System.out.println("我是主线程, 执行另外的业务...");

        System.out.println("task执行结果：" + futureTask.get(5, TimeUnit.SECONDS));

        System.out.println("任务全部执行完毕...");
    }
}
class Task implements Callable<String> {

    @Override
    public String call() throws Exception {
        System.out.println("子线程, 业务处理中...");
        Thread.sleep(2000);
        return "业务执行成功";
    }
}

View Code

shiro中session验证定时任务

　　shiro源码篇 - shiro的session的查询、刷新、过期与删除，你值得拥有中讲到了session验证定时任务，我们AbstractValidatingSessionManager中createSession方法开始

　　可以看到，调用Executors.newSingleThreadScheduledExcutor(ThreadFactory threadFactory)方法创建了一个支持定时及周期性执行的单线程化线程池，支持定时及周期性地执行task，并且线程池中只有一个线程。ExecutorServiceSessionValidationScheduler本身就是一个Runnable，那么会定时、周期性的执行其run()。说的简单点就是：应用启动60分钟后，单线程化的线程池中的单个线程开始执行ExecutorServiceSessionValidationScheduler的run()方法，之后每隔60分钟执行一次，60分钟是默认设置；ExecutorServiceSessionValidationScheduler的run()中，会调用sessionManager的validateSessions()方法完成session的验证。

总结

　　1、无需返回结果，简单的线程实现可以用Runnable（或Thread）；需要返回结果的、稍复杂的线程实现可以用Callable；如果线程操作频繁、需要连接池管理的可以考虑用ExecutorService来实现线程池；更复杂的任务调度，则可以用三方工具，比如：quartz，更多三方调度工具可查阅spring-boot-2.0.3之quartz集成，不是你想的那样哦！，具体选择哪个，需要结合我们的具体业务来考虑，没有绝对的选择谁而不选择谁，就看谁更契合；

　　2、一般情况下，Callable(或Runnale)、Executors、ExecutorService、Future会配合来使用，很多时候我们不需要返回值，则可以不关注Future；推荐使用线程池的方式，有与数据库连接池类似的优点；

　　3、很多三方的框架、工具都沿用了jdk的线程池实现，而没有引用第三方调度工具，例如shiro中，session的验证定时任务就是沿用的jdk中的Executors.newSingleThreadScheduledExcutor(ThreadFactory threadFactory)来创建的线程池；

　　4、jdk中的线程还有很多内容，本文只是涉及到了冰山一角，更深入的学习有待大家自行去进行。

参考

　　Java 8 教程汇总

　　Java并发编程：Callable、Future和FutureTask

　　深入理解 Java 线程池：ThreadPoolExecutor