Java中的线程池和异步任务详解_java 线程池异步提交

Java中的线程池和异步任务详解

转载自：https://blog.csdn.net/fanrenxiang/article/details/79855992
引言
java中经常需要用到多线程来处理一些业务，我们非常不建议单纯使用继承Thread或者实现Runnable接口的方式来创建线程，那样势必有创建及销毁线程耗费资源、线程上下文切换问题，这个时候引入线程池比较合理。有些时候也需要把多线程的逻辑给异步话，接口不需要等待子线程逻辑执行完就马上返回，这里就需要异步任务。java中涉及到线程池的相关类均在jdk1.5开始的java.util.concurrent包中，涉及到的几个核心类及接口包括：ExecutorService、Executors、ThreadPoolExecutor、FeatureTask、Callable、Runnable等。后面会一一描述。
伪代码思路如下：

 @ResponseBody
    @RequestMapping(value = "/notifyOrder", method = RequestMethod.POST)
    public CommonResult notifyOrder(@RequestParam(value = "req", required = false) String req) {
        log.info("notifyOrder method parameter：req={}", req);
        CommonResult result = new CommonResult();
        // 1、校验工作
 
        //  2、decode送过来的原始数据并序列化
        PointsMallOrderInfo orderInfo = StringToBean(req, "data", PointsMallOrderInfo.class);
 
        // 3、保存推送过来的原始数据
        orderInfoService.addOrderInfo(orderInfo);
 
        // 4、获取系统中状态为可用的虚拟码
 
        // 5、多线程异步处理 发送短信、提交虚拟码
        AsynSendAndSubmitTask.asynSendAndSubmit(orderInfo);
 
        // 6、持久化虚拟码流水号等信息
        return result;
    }
 
public class AsynSendAndSubmitTask {
    private static ExecutorService asynSendAndSubmitThreadPool = new ThreadPoolExecutor(10, 10, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(40));
 
    public static void asynSendAndSubmit(final PointsMallOrderInfo orderInfo) {
        asynSendAndSubmitThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                // 短信下发虚拟码给用户
                Long messageId = smsService.sendSms(new SendSmsInfo());
                // 给积分商城提交虚拟码
                String submitResult = HttpUtil.httpPost(false, params, SETVIRTUALCODE_URL);
            }
        });
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

我们程序逻辑主线程中执行“保存推送过来的基础数据、查询并取出可用的充值码、持久化充值码及相关流水号信息操作，但提交虚拟码和短信发送虚拟码则以多线程的方式异步处理，加快接口响应。（当然也可以用kafka/MQ之类的消息队列替代）

角色划分
1、任务：指的是实现了Callable或Runnable接口的类，里面包含主要的业务逻辑，任务用于提交至线程池里的线程去执行。实现Callable接口的任务类可以有返回值，而Runnable接口则无；
2、异步计算结果：主要是FutureTask类，里面包含了异步任务的计算结果，可以理解为执行单元，用于提交至线程池时使用；
3、线程池：ThreadPoolExecutor和Executors类，用于构建线程池。
Executors 和ThreadPoolExecutor
由源码可知Executors本质上还是使用的ThreadPoolExecutor来实例化线程池的，Executors类可以创建四种类型的线程池，分别为 newFixedThreadPool、newSingleThreadExecutor、newCachedThreadPool、newScheduledThreadPool
FixedThreadPool
使用固定线程数，适用于为了平衡服务器资源而指定线程数的场景，一般用于负载比较高的服务器

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>(),
                                  threadFactory);
}
1
2
3
4
5
6
7
8
9
10
11

如上是在ThreadPoolExecutor类中的源码，可以看到newFixedThreadPool线程池重用放在"共享无边界的队列LinkedBlockingQueue"中的固定线程数，处理任务时，nThread大部分都将以活动状态在处理任务，如果当所有线程都处于活动状态时又有额外的任务提交，那么新任务将在队列中等待直至线程可用。
SingleThreadExecutor
使用单个线程数，适用于需要保证顺序的执行各个任务；并且在任意时间点，不会有多个线程活动的场景

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>(),
                                threadFactory));
}
1
2
3
4
5
6
7
8
9
10
11
12
13

关于这种方式创建的线程池，源码中特别强调了：如果该单个线程在关闭之前由于执行过程中的失败而终止，那么如果需要执行后续任务，将替换一个新的线程，使用到的也是LinkedBlockingQueue队列。
CachedThreadPool
使用非固定线程数，适用于执行很多短期的异步任务，或者负载较轻的服务器

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>(),
                                  threadFactory);
}
1
2
3
4
5
6
7
8
9
10
11

根据需要创建线程，如优重用先前创建的可用线程，否则将新建线程并放到池子中，从构造函数可以看到，默认超过60秒未使用的线程将被终止并从cache中移除，使用到的同步移交SynchronousQueue队列。
ScheduledThreadPool
包含多个线程，适用于需要多个后台执行周期任务（或延迟任务），同时为了满足资源管理的需求而需要限制线程数量的场景

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}
public static ScheduledExecutorService newScheduledThreadPool(
        int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}
1
2
3
4
5
6
7

对应的还有个SingleThreadScheduledExecutor只包含单个线程，适用于需要单个后台线程执行周期性任务，同时需要保证顺序的执行各个任务的场景

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1));
}
public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1, threadFactory));
}
1
2
3
4
5
6
7
8

使用Executors创建线程池的隐患
我这里以newFixedThreadPool构建的线程池为例，结合源码看看Executors.newFixedThreadPool(n)创建的线程池会有哪些潜在的隐患。进入newFixedThreadPool初始化的源码

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {……}
1
2
3
4
5
6
7

corePoolSize：线程池核心线程数，要注意线程池初创时候并不会启动corePoolSize个线程，而是随着任务的提交逐渐达到这个值；
maximumPoolSize：池中的最大线程数，要注意这个参数只有在任务数量大于corePoolSize时才会起作用；
keepAliveTime：当线程数大于corePoolSize时，多余空闲线程在终止之前等待新任务的最大时间；
unit：keepAliveTime的时间单位；
workQueue：用于保存任务的队列，可以为无界、有界、同步移交类型的队列，这里是BlockingQueue。当池子里的工作线程数大于corePoolSize时，这时新进来的任务才会放到阻塞队列中，；
threadFactory：创建新的线程时的工厂类，比如guava的ThreadFactoryBuilder；
handler：队列已满且线程数达到maximunPoolSize时候的饱和策略，取值有AbortPolicy、CallerRunsPolicy、DiscardOldestPolicy、DiscardPolicy；
上面说到了 FixedThreadPool、SingleThreadExecutor 源码中使用到的是LinkedBlockingQueue无界队列，而CacheThreadPool和ScheduledThreadPool实例化时默认最大线程数又是Integer.MAX_VALUE，这可能导致什么结果呢？《阿里巴巴Java开发手册v1.2.0》中这样说到：
结合源码看：FixedThreadPool、SingleThreadExecutor的LinkedBlockQueue是一个用链表实现的有界阻塞队列，容量可以选择进行设置，默认将是一个无边界的阻塞队列，最大长度为Integer.MAX_VALUE.

  /**
   * Creates a {@code LinkedBlockingQueue} with a capacity of
   * {@link Integer#MAX_VALUE}.
   */
  public LinkedBlockingQueue() {
      this(Integer.MAX_VALUE);
  }
1
2
3
4
5
6
7

而CacheThreadPool和ScheduledThreadPool实例化时默认最大允许创建的线程数是Integer.MAX_VALUE

 public static ExecutorService newCachedThreadPool() {
     return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                   60L, TimeUnit.SECONDS,
                                   new SynchronousQueue<Runnable>());
 }
 
 public ScheduledThreadPoolExecutor(int corePoolSize) {
     super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
           new DelayedWorkQueue());
 }
1
2
3
4
5
6
7
8
9
10

所以上述两个"漏洞"在特定的场景下就有可能会导致OOM，故而很多人都不建议使用这颗"定时炸弹"。
创建线程池的正确姿势？
那么上面说了使用Executors创建的线程池有隐患，那如何使用才能避免这个隐患呢？对症下药，既然FixedThreadPool和SingleThreadPool"可能"导致的OOM是由于使用了无界队列任务堆积，CacheThreadPool和ScheduledThreadPool是由于"可能"创建Interger.MAX_VALUE，那创建线程池时我们就使用有界队列或者指定最大允许创建线程个数即可。使用下面的构造函数

private static ExecutorService executor = new ThreadPoolExecutor(10,10,60L, TimeUnit.SECONDS,new ArrayBlockingQueue(10));
1

这样可以指定corePoolSize、maximumPoolSize、workQueue为ArrayBlockingQueue有界队列

 public ThreadPoolExecutor(int corePoolSize,
                           int maximumPoolSize,
                           long keepAliveTime,
                           TimeUnit unit,
                           BlockingQueue<Runnable> workQueue) {
     this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
          Executors.defaultThreadFactory(), defaultHandler);
 }
 private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
1
2
3
4
5
6
7
8
9

默认的handler队列饱和策略为AbortPolicy（直接抛出异常），当提交任务线程数高于可用线程数，队列放满而无法处理新请求，这时候会抛出java.util.concurrent.RejectedExecutionException异常，然后手动捕获即可，总比OOM强吧。当然你也可以使用guava包中的ThreadFactoryBuilder工厂类来构造线程池:

private static ThreadFactory threadFactory = new ThreadFactoryBuilder().build();
private static ExecutorService executorService = new ThreadPoolExecutor(10, 10, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10), threadFactory, new ThreadPoolExecutor.AbortPolicy());
1
2

通过guava的ThreadFactory工厂类还可以指定线程的名称，这对于后期定位错误时也是很有帮助的

ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("thread-pool-d%").build();
1

Future和FutureTask（异步任务结果）
Future接口和FutureTask类用来表示执行异步任务的结果，当向ThreadPoolExecutor或ScheduledThreadPoolExecutor提交了一个Callable或Runnable接口的实现类时，ThreadPoolExecutor或ScheduledThreadPoolExecutor就会返回FutureTask。到目前的jdk版本为止，submit返回的是都是实现了Future接口的FutureTask。

<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
1
2
3

FutureTask一般都是和线程池搭配使用，用于多线程的方式提交任务，通过futureTask.get()方法获取异步任务的计算结果即可。如下代码:

import java.util.concurrent.*;
 
public class FutureTaskTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executor = Executors.newCachedThreadPool();
        try {
            CustomCallable callable = new CustomCallable();
            FutureTask<Integer> callableTask = new FutureTask<Integer>(callable);
            executor.submit(callableTask);
            System.out.println("callableTask任务计算结果=" + callableTask.get());
 
            CustomRunnable runnable = new CustomRunnable();
            FutureTask<Void> runnableTask = new FutureTask<>(runnable, null);
            executor.submit(runnableTask);
 
        } finally {
            executor.shutdown();
        }
    }
}
 
class CustomCallable implements Callable<Integer> {
    @Override
    public Integer call() throws Exception {
        int sum = 0;
        for (int i = 0; i < 10; i++) {
            sum += i;
        }
        return sum;
    }
}
 
class CustomRunnable implements Runnable {
    @Override
    public void run() {
        int sum = 0;
        for (int i = 0; i < 10; i++) {
            sum += i;
        }
        System.out.println("RunnableTask任务计算结果=" + sum);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

从上面代码和开头讲到的demo1场景来看，使用FutureTask后，提交给线程池的就变成了futuretask而不是简单的实现了Runnable或Callable接口的普通任务了，并且获取任务的结果也是通过futuretask.get()方法而不是executorservice.submit()返回值。
Callable和Runnable（任务类逻辑）
Callable接口和Runnable接口的实现类（近似等价于被提交任务的逻辑）均可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor所执行，区别在于执行的任务逻辑是否需要返回值，Callable接口实现类可以有返回值，而Runnable接口实现类则没有返回值；比如这里自定义一个实现了Callable接口的任务类：

class CustomCallable implements Callable<Integer> {
    @Override
    public Integer call() throws Exception {
        System.out.println("子线程开始进行计算");
        Thread.sleep(500);
        int sum = 0;
        for (int i = 0; i < 10; i++)
            sum += i;
        return sum;
    }
}
1
2
3
4
5
6
7
8
9
10
11

也可以通过Executors类包装的如下两种方式创建任务

//此方式创建Callable对象，通过futureTask.get()方法可以获取到异步计算结果
public static <T> Callable<T> callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<T>(task, result);
}
//不返回异步计算结果
public static Callable<Object> callable(Runnable task) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<Object>(task, null);
}
1
2
3
4
5
6
7
8
9
10
11
12

1、通过futureTask.get()方法获取任务计算结果时，当任务还未完成，会导致线程阻塞直至任务完成，一般会配合futureTask.isDone()方法判断子线程任务是否完成来一起使用;

2、当项目中有很多异步任务时，要着重测试下每个异步任务的执行时间，比如某个异步任务是调用其他系统的web服务，这时候就得测试调用需用的时间长短，如果过长，则建议使用生产/消费模式的消息队列去实现，不然容易使服务器的jvm进程崩溃;

线程池实现原理？
current包下的线程池实现原理相对简单，就是一个线程集合workers和存放任务的阻塞队列workQueue，当有新任务提交时就放到workQueue队列中（注:当池子里的核心线程数小于corePoolSize时任务会直接被执行），然后线程池从任务阻塞队列中"分配"任务并执行。

/**
  * 设置线程池中的所有工作线程
  */
 private final HashSet<Worker> workers = new HashSet<Worker>();
 /**
  * 用于保存任务和切换到工作线程的队列
  */
 private final BlockingQueue<Runnable> workQueue;
1
2
3
4
5
6
7
8

Springboot中使用线程池
springboot可以说是非常流行了，下面说说如何在springboot中让spring来帮我们管理线程池

/**
 * @ClassName ThreadPoolConfig
 * @Description 构建spring管理的线程池实例，方便调用
 * @Author simonsfan
 * @Date 2018/12/20
 * Version  1.0
 */
@Configuration
public class ThreadPoolConfig {
 
    @Bean(value = "threadPoolInstance")
    public ExecutorService createThreadPoolInstance() {
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("thread-pool-%d").build();
        ExecutorService threadPool = new ThreadPoolExecutor(10, 10, 30L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10), threadFactory, new ThreadPoolExecutor.AbortPolicy());
        return threadPool;
    }
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

 @Resource(name = "threadPoolInstance")
    private ExecutorService executorService;
 
    @Override
    public void spikeConsumer() {
        //TODO
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                //TODO
            }
        });
    }
1
2
3
4
5
6
7
8
9
10
11
12
13

总结

这篇文章主要讲的是线程池的知识，但是还是比较的基础的，如果想继续研究，还需要看一些其它的文章。