Java多线程入门（六）——Callable、Future和线程池

       本文介绍另外两种创建多线程的方式，这两种方式我们在实际中会用的多一点，尤其是线程池。而在前面文章中我们讲述了创建线程最基本的两种方式：一种是直接继承Thread，另外一种就是实现Runnable接口。但是这两种方式创建线程有一个缺陷，那就是无法获取到线程运行后的结果，因为这两个方式都是重写了 run（）方法，而run（）方法是用void修饰的。所以后来就有了Callable和Future这两个接口，它们能够获取线程执行的结果。

1、Callable

    Callable是在JDK1.5中出现的接口，它和Runnable接口很相似，我们可以认为：Callable接口是Runnable接口的增强版，因为Runnable有的功能Callable都有，而且还能获取任务执行的结果。所以下面我们来看一下Callable和Runnable接口的对比：

     先来看一下Runnable接口的源码：

public interface Runnable {

    public abstract void run();

}

    Callable接口的源代码：

public interface Callable<V> {

    V call() throws Exception;

}

   可以很明显的看出它们二者的区别：

1. Callable使用的是call（），而Runnable中使用的是run（）。
2. Callable的call（）可以抛出异常，而Runnable的run（）不会抛出异常。
3. Callable能接受一个泛型，然后在call（）中返回一个这个类型的值。而Runnable的run（）没有返回值。

我们上面说Callable是可以返回任务执行结果的，而获取返回结果需使用到Future。所以下面要介绍一下Future。

2、Future

    Future也是一个接口，通过它可以获得任务执行的返回值。该接口的内部源码如下：

public interface Future<V> {
　　boolean cancel(boolean mayInterruptIfRunning);

　　boolean isCancelled();

　　boolean isDone();

　　V get() throws InterruptedException, ExecutionException;

　　V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

可以发现在Future接口中声明了5个方法，下面依次解释每个方法的作用：

• cancel方法用来取消任务，如果取消任务成功则返回true，如果取消任务失败则返回false。参数mayInterruptIfRunning表示是否允许取消正在执行却没有执行完毕的任务，如果设置true，则表示可以取消正在执行过程中的任务。如果任务已经完成，则无论mayInterruptIfRunning为true还是false，此方法肯定返回false，即如果取消已经完成的任务会返回false；如果任务正在执行，若mayInterruptIfRunning设置为true，则返回true，若mayInterruptIfRunning设置为false，则返回false；如果任务还没有执行，则无论mayInterruptIfRunning为true还是false，肯定返回true。
• isCancelled方法表示任务是否被取消成功，如果在任务正常完成前被取消成功，则返回 true。
• isDone方法表示任务是否已经完成，若任务完成，则返回true；
• get()方法用来获取执行结果，这个方法会产生阻塞，会一直等到任务执行完毕才返回；
• get(long timeout, TimeUnit unit)用来获取执行结果，如果在指定时间内，还没获取到结果，就会抛出TimeoutException异常（慎用这个方法，因为有很多坑）。

     

        下面是Future接口中在java.util.concurrent包下类的结构图：

由于Future只是一个接口，所以是无法直接用来创建对象使用的，所以真正获取结果用到的是FutureTask这个类。

   FutureTask

通过上面的图片发现FutureTask类是实现了RunnableFuture接口，而这个接口又继承了Future接口，我们具体点开其源码来看。

public class FutureTask<V> implements RunnableFuture<V>{

	code...
}

打开RunnableFuture接口的实现：

public interface RunnableFuture<V> extends Runnable, Future<V> {

    void run();

}

可以看出RunnableFuture继承了Runnable接口和Future接口，而FutureTask实现了RunnableFuture接口。所以它既可以作为Runnable被线程执行，又可以作为Future得到Callable的返回值。

 

   上面说了这么多，接下来使用Callable+FutureTask创建线程并获取执行结果的一个栗子如下：

1. 创建一个实现Callable接口的类。
2. 重写call方法，将线程要执行的操作定义在call()中。
3. 创建Callable接口实现类的对象。
4. 创建FutureTask对象，并将上面Callable接口实现类的对象传入FutureTask构造器中。
5. 将FutureTask的对象作为参数传入Thread类的构造器中，创建Thread类对象，并且启动线程。
6. 获取Callable中call方法的返回值。

package com.thr;

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

/**
 * @author Administrator
 * @date 2020-04-09
 * @desc Callable+Future创建并获取线程执行结果
 */
//1、创建一个实现Callable接口的类
class MyCallable implements Callable<Integer>{

    //2、重写call方法，将线程要执行的操作定义在call()中
    @Override
    public Integer call() throws Exception {
        int num=0;
        for (int i = 1; i <= 100; i++) {
            num+=i;
        }
        return num;
    }
}

public class CallableFutureDemo {

    public static void main(String[] args) {
        //3、创建Callable接口实现类的对象
        MyCallable callable = new MyCallable();
        //4、创建FutureTask对象，并将上面Callable接口实现类的对象传入FutureTask构造器中
        FutureTask<Integer> task = new FutureTask<Integer>(callable);
        //5、将FutureTask的对象作为参数传入Thread类的构造器中，创建Thread类对象，并且启动线程
        new Thread(task).start();
        try {
            //6、获取Callable中call方法的返回值
            Integer integer = task.get();
            System.out.println(integer);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}

 

    我们知道Callable用于产生结果，Future用于获取结果。不过Callable和Future一般都和线程池搭配使用，所以下面再来简单介绍一下线程池的使用。

3、线程池

    在前面的文章中我们介绍了Thread、Runnable和Callable这三种方式创建线程，我们在创建少量线程的时候使用它们是非常的简单方便的，但是如果我们需要创建成百上千的线程时，那么岂不是要创建成百上千个线程对象，调用成百上千的start()方法，可见这样是非常浪费时间、消耗资源和降低程序效率的。那么为了解决这一问题就出现了线程池。

    线程池顾名思义，就是由很多线程构成的池子。在有任务的时候随时取用线程，当任务完成后又将线程放回池中。

    所以合理利用线程池能够带来三个好处。

1. 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
2. 提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
3. 提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

   线程池的创建一般用Executors这个工具类来创建，常见的有以下四种方式：

• newFixedThreadPool(int nThreads):创建一固定线程数目的线程池，超出的线程会在队列中等待。
• newSingleThreadPoolExecutor():创建一个单线程化的线程池。它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
• newCacheThreadPool():创建一个可缓存的线程池。如果现有线程没有可用的，则创建一个新线程并添加到缓存池中。如果有被使用完但是还没销毁的线程，就复用该线程。如果有线程60s未被使用的话就会从缓存中移出并终止。因此，长时间保持空闲的线程池不会使用任何资源。
• newScheduledThreadPool(int corePoolSize)：创建一个支持定时及周期性的任务执行的线程池，多数情况下可用来替代Time类。

  但是并不推荐使用Executors来创建，因为可能会出现OOM（Out Of Memory，内存溢出）的情况，下面我们依次详细的分析这四个方式：

    ①、Executors.newFixedThreadPool(int nThread)

 public static ExecutorService newFixedThreadPool(int nThreads) {
       return new ThreadPoolExecutor(nThreads, nThreads,
                                     0L, TimeUnit.MILLISECONDS,
                               new LinkedBlockingQueue<Runnable>());
 }

    可以发现最后一行使用了LinkedBlockingQueue，泛型是Runnable类型，这里的队列是用来存放线程任务的。我们再来看看这个LinkedBlockingQueue部分源码：

public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
}

public LinkedBlockingQueue(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
}

     在上一章博客中提过LinkedBlockingQueue是链表实现的有界阻塞队列，其capacity是可以选择进行设置的，如果不设置的话，将是一个无边界的阻塞队列，队列的最大长度为Integer.MAX_VALUE。而上面newFixedThreadPool的源码中，我们可以很清晰的看到LinkedBlockingQueue是没有指定capacity的。所以此时LinkedBlockingQueue就是一个无边界队列，对于一个无边界队列来说，是可以不断的向队列中加入任务的，这种情况下就有可能因为队列中等待的线程数太多而导致OOM。

    下面我们来一个简单的例子，模拟一下使用Executors导致OOM的情况：

    首先将JVM参数调一下：-Xmx8m –Xms8m

    代码如下：

package com.thr;

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

/**
 * @author Administrator
 * @date 2020-04-11
 * @desc Excutors出现OOM举例
 */
public class ExecutorsDemo {
    private static ExecutorService service = Executors.newFixedThreadPool(15);

    public static void main(String[] args) {
        for (int i = 0; i < Integer.MAX_VALUE; i++) {
            service.execute(new SubThread());
        }
    }
}

class SubThread implements Runnable {
    @Override
    public void run() {
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            //do nothing
        }
    }
}

    运行结果：

②、Executors.newSingleThreadExexutor()

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
}

可以发现还是使用阻塞队列LinkedBlockingQueue，所以问题是一样的。

③、Executors.newCacheThreadPool()

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
}

可以发现ThreadPoolExecutor对象中的第二个参数为Integer.MAX_VALUE，而这个位置参数的意思为线程池最大线程数。所以还是会出现OOM的情况。

④、Executors.newScheduleThreadPool()

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
}

public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {

		code...

		}

    通过上面三段代码可以发现newScheduleThreadPool()方法返回了ScheduledThreadPoolExecutor对象，而它又继承了ThreadPoolExecutor类，并且调用的是父类的构造器，而构造器中的第二个参数为Integer.MAX_VALUE，所以还是同样的问题。

    这就是使用Executors工具类创建线程池的缺陷所在，在《阿里巴巴开发手册》中是不建议使用这种方式创建线程池的，而是推荐使用new ThreadPoolExecutor构造函数来创建线程池。如果你细心一点会发现上面四种方式中其实最终都是使用ThreadPoolExecutor这个类，所以这个类才是线程池的核心，我们只有彻底了解这个类才能真正的理解线程池。

 

4、ThreadPoolExecutor

       上面既然说推荐使用ThreadPoolExecutor来创建线程池，那么先来看一下ThreadPoolExecutor的内容。在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;
    }

   可以发现构造器有7个参数，下面分别解释下构造器中各个参数的含义：

• corePoolSize：核心池的大小。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这2个方法是预创建线程的意思，即在没有任务到来之前就创建corePoolSize个线程或者一个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中。
• maximumPoolSize：线程池最大线程数，这个参数也是一个非常重要的参数，它表示在线程池中最多能创建多少个线程。
• keepAliveTime：空闲线程的存活时间。就是当线程的数量大于corePoolSize时，如果等待了keepAliveTime时长还没有任务可执行，则线程终止（前提是线程池中的线程数必须大于corePoolSize时，keepAliveTime才会起作用），直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0。
• unit：参数keepAliveTime的时间单位，它在TimeUnit类中有7种静态属性可取。

TimeUnit.DAYS;              //天
TimeUnit.HOURS;             //小时
TimeUnit.MINUTES;           //分钟
TimeUnit.SECONDS;           //秒
TimeUnit.MILLISECONDS;      //毫秒
TimeUnit.MICROSECONDS;      //微妙
TimeUnit.NANOSECONDS;       //纳秒

• workQueue：一个阻塞队列，用来存储等待执行的任务，这个参数的选择也很重要，会对线程池的运行过程产生重大影响，一般来说，这里的阻塞队列有以下几种选择：

ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按FIFO（先进先出）原则对元素进行排序。
LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO排序元素，吞吐量通常要高于ArrayBlockingQueue，静态工厂方法Executors.newFixedThreadPool()使用了这个队列。
SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个列。
PriorityBlockingQueue：一个具有优先级的无限阻塞队列。
其中ArrayBlockingQueue和PriorityBlockingQueue使用较少，一般使用LinkedBlockingQueue和SynchronousQueue。

• threadFactory：线程工厂。用于创建新线程。threadFactory创建的线程也是采用new Thread()方式，threadFactory创建的线程名都具有统一的风格：pool-m-thread-n（m为线程池的编号，n为线程池内的线程编号）。
• handler：线程饱和策略。当线程池和队列都满了，再加入线程会执行此策略，它有四种策略。

ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

上面既然介绍完了线程池构造方法中的各个参数，那么再来介绍线程池的工作流程：

1）当提交一个新任务到线程池时，线程池判断corePoolSize线程池是否都在执行任务，如果有空闲线程，则创建一个新的工作线程来执行任务，直到当前线程数等于corePoolSize；

2）如果当前线程数为corePoolSize，继续提交的任务被保存到阻塞队列中，等待被执行；

3）如果阻塞队列满了，那就创建新的线程执行当前任务，直到线程池中的线程数达到maxPoolSize，这时再有任务来，由饱和策略来处理提交的任务

我们在打开ThreadPoolExecutor类的代码可以看到，ThreadPoolExecutor继承了AbstractExecutorService，我们来看一下AbstractExecutorService的实现：

public abstract class AbstractExecutorService implements ExecutorService {


    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { };
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { };
    public Future<?> submit(Runnable task) {};
    public <T> Future<T> submit(Runnable task, T result) { };
    public <T> Future<T> submit(Callable<T> task) { };
    private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,boolean timed, long nanos)
        throws InterruptedException, ExecutionException, TimeoutException {
    };
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException {
    };
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
    };
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {
    };
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
        throws InterruptedException {
    };
}

　AbstractExecutorService是一个抽象类，它实现了ExecutorService接口。我们接着看ExecutorService接口的实现：

public interface ExecutorService extends Executor {

    void shutdown();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;
    <T> Future<T> submit(Callable<T> task);
    <T> Future<T> submit(Runnable task, T result);
    Future<?> submit(Runnable task);
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)
        throws InterruptedException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;
    <T> T invokeAny(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

　而ExecutorService又是继承了Executor接口，我们看一下Executor接口的实现：

public interface Executor {

    void execute(Runnable command);
}

    所以到这里，大家应该明白了ThreadPoolExecutor、AbstractExecutorService、ExecutorService和Executor几个之间的关系了。

    Executor是一个线程池顶层接口，在它里面只声明了一个方法execute(Runnable)，返回值为void，参数为Runnable类型，它就是用来执行传进去的任务的，但没有返回值；

    然后ExecutorService接口继承了Executor接口，并声明了一些方法：submit、invokeAll、invokeAny以及shutdown等；

    抽象类AbstractExecutorService实现了ExecutorService接口，基本实现了ExecutorService中声明的所有方法；

    然后ThreadPoolExecutor继承了类AbstractExecutorService。

    在ThreadPoolExecutor类中有几个非常重要的方法：

• execute(Runnable command)：用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功。
• submit(Callable<T> task)/submit(Runnable task)：用于提交需要返回值的任务。线程池会返回一个future类型的对象，通过这个future对象可以判断任务是否执行成功，并且可以通过future的get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用get(long timeout，TimeUnit unit)，在指定的时间内会等待任务执行，超时则抛出超时异常，等待时候会阻塞当前线程。
• shutdown()：不会立即终止线程池，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务。
• awaitTermination(long timeout, TimeUnit unit)：用于设定超时时间及单位。当等待超过设定时间时，会监测ExecutorService是否已经关闭，若关闭则返回true，否则返回false。一般情况下会和shutdown方法组合使用。
• shutdownNow()：立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务。

    所以通过上面的讲解大家应该知道创建线程池的正确姿势了吧：

    ExecutorService es = new ThreadPoolExecutor(5, 20,0L,
            TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(10));

    简单的举个例子吧：

package com.thr;

import java.util.concurrent.*;

/**
 * @author Administrator
 * @date 2020-04-11
 * @desc 使用自定义参数ThreadPoolExecutor创建线程池
 */
public class ExecutorServiceDemo {
    public static void main(String[] args) {
    //定义线程池参数
        ExecutorService es = new ThreadPoolExecutor(5, 20,0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(10));
        //创建Callable和Future对象
        MyCallable myCallable = new MyCallable();
        Future<Integer> future = es.submit(myCallable);
        try {
            //获取结果并打印
            Integer num = future.get();
            System.out.println(num);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }finally {
            //关闭线程池
            es.shutdown();
        }
    }
}

class MyCallable implements Callable<Integer>{

    @Override
    public Integer call() throws Exception {
        int sum=0;
        for (int i = 1; i <= 100; i++) {
            sum+=i;
        }
        return sum;
    }
}

       当然除了自己定义ThreadPoolExecutor外。还有其他方法。比如各种开源工具如Guava等。这里推荐使用Guava提供的ThreadFactoryBuilder来创建线程池。因为当我们需要给新创建的线程取名字、或者设置为守护线程、错误处理器等操作时，它的好处就体现出来了。简单举例：(注意使用Guava需要引入包)

public class ThreadFactoryBuilderTest {

    public static void main(String[] args) {
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("线程名称-%s").build();
        // 创建一个线程对象
        Thread newThread = threadFactory.newThread(()->{

        });
        System.out.println(newThread.getName());
    }
}

参考资料：

https://www.cnblogs.com/dolphin0520/p/3949310.html

https://www.cnblogs.com/dolphin0520/p/3932921.html

https://blog.csdn.net/hollis_chuang/article/details/83743723

https://zhuanlan.zhihu.com/p/32867181