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

Java基础

浏览数:1,115

2019-3-26

前言

  开心一刻

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

纯正的海豹突击队

  路漫漫其修远兮,吾将上下而求索!

  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();

    }
}

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

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

  当然线程的实现方式还有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;
}

    从如上代码可以看出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");
    }
}

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

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

  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 "业务执行成功";
    }
}

    示例二,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 "业务执行成功";
    }
}

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