⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处苏三说技术「苏三」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

前言

最近有位小伙伴在我的技术群里，问了我一个问题：服务down机了，线程池中如何保证不丢失数据？

这个问题挺有意思的，今天通过这篇文章，拿出来跟大家一起探讨一下。

1 什么是线程池？

之前没有线程池的时候，我们在代码中，创建一个线程有两种方式：

继承Thread类
实现Runnable接口

虽说通过这两种方式创建一个线程，非常方便。

但也带来了下面的问题：

创建和销毁一个线程，都是比较耗时，频繁的创建和销毁线程，非常影响系统的性能。
无限制的创建线程，会导致内存不足。
有新任务过来时，必须要先创建好线程才能执行，不能直接复用线程。

为了解决上面的这些问题，Java中引入了：线程池。

它相当于一个存放线程的池子。

使用线程池带来了下面3个好处：

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

2 线程池原理

先看看线程池的构造器：

public ThreadPoolExecutor(
    int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    ThreadFactory threadFactory,
    RejectedExecutionHandler handler)

corePoolSize：核心线程数，线程池维护的最少线程数。
maximumPoolSize：最大线程数，线程池允许创建的最大线程数。
keepAliveTime：线程存活时间，当线程数超过核心线程数时，多余的空闲线程的存活时间。
unit：时间单位。
workQueue：任务队列，用于保存等待执行的任务。
threadFactory：线程工厂，用于创建新线程。
handler：拒绝策略，当任务无法执行时的处理策略。

线程池的核心流程图如下：

线程池的工作过程如下：

线程池初始化：根据corePoolSize初始化核心线程。
任务提交：当任务提交到线程池时，根据当前线程数判断：

若当前线程数小于corePoolSize，创建新的线程执行任务。
若当前线程数大于或等于corePoolSize，任务被加入workQueue队列。

任务处理：当有空闲线程时，从workQueue中取出任务执行。
线程扩展：若队列已满且当前线程数小于maximumPoolSize，创建新的线程处理任务。
线程回收：当线程空闲时间超过keepAliveTime，多余的线程会被回收，直到线程数不超过corePoolSize。
拒绝策略：若队列已满且当前线程数达到maximumPoolSize，则根据拒绝策略处理新任务。

说白了在线程池中，多余的任务会被放到workQueue任务队列中。

这个任务队列的数据保存在内存中。

这样就会出现一些问题。

接下来，看看线程池有哪些问题。

3 线程池有哪些问题？

在JDK中为了方便大家创建线程池，专门提供了Executors这个工具类。

3.1 队列过大

Executors.newFixedThreadPool，它可以创建固定线程数量的线程池，任务队列使用的是LinkedBlockingQueue，默认最大容量是Integer.MAX_VALUE。

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

如果向newFixedThreadPool线程池中提交的任务太多，可能会导致LinkedBlockingQueue非常大，从而出现OOM问题。

3.2 线程太多

Executors.newCachedThreadPool，它可以创建可缓冲的线程池，最大线程数量是Integer.MAX_VALUE，任务队列使用的是SynchronousQueue。

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

如果向newCachedThreadPool线程池中提交的任务太多，可能会导致创建大量的线程，也会出现OOM问题。

3.3 数据丢失

如果线程池在执行过程中，服务突然被重启了，可能会导致线程池中的数据丢失。

上面的OOM问题，我们在日常开发中，可以通过自定义线程池的方式解决。

比如创建这样的线程池：

new ThreadPoolExecutor(8, 
                       10,
                       30L, 
     TimeUnit.MILLISECONDS,
    new ArrayBlockingQueue<Runnable>(300),
            threadFactory);

自定义了一个最大线程数量和任务队列都在可控范围内线程池。

这样做基本上不会出现OOM问题。

但线程池的数据丢失问题，光靠自身的功能很难解决。

4 如何保证数据不丢失？

线程池中的数据，是保存到内存中的，一旦遇到服务器重启了，数据就会丢失。

之前的系统流程是这样的：

用户请求过来之后，先处理业务逻辑1，它是系统的核心功能。

然后再将任务提交到线程池，由它处理业务逻辑2，它是系统的非核心功能。

但如果线程池在处理的过程中，服务down机了，此时，业务逻辑2的数据就会丢失。

那么，如何保证数据不丢失呢？

答：需要提前做持久化。

我们优化的系统流程如下：

用户请求过来之后，先处理业务逻辑1，紧接着向DB中写入一条任务数据，状态是：待执行。

处理业务逻辑1和向DB写任务数据，可以在同一个事务中，方便出现异常时回滚。

然后有一个专门的定时任务，每个一段时间，按添加时间升序，分页查询状态是待执行的任务。

最早的任务，最先被查出来。

然后将查出的任务提交到线程池中，由它处理业务逻辑2。

处理成功之后，修改任务的待执行状态为：已执行。

需要注意的是：业务逻辑2的处理过程，要做幂等性设计，同一个请求允许被执行多次，其结果不会有影响。

如果此时，线程池在处理的过程中，服务down机了，业务逻辑2的数据会丢失。

但此时DB中保存了任务的数据，并且丢失那些任务的状态还是：待执行。

在下一次定时任务周期开始执行时，又会将那些任务数据重新查询出来，重新提交到线程池中。

业务逻辑2丢失的数据，又自动回来了。

如果要考虑失败的情况，还需要在任务表中增加一个失败次数字段。

在定时任务的线程池中执行业务逻辑2失败了，在下定时任务执行时可以自动重试。

但不可能无限制的一直重试下去。

当失败超过了一定的次数，可以将任务状态改成：失败。

这样后续可以人工处理。

芋道源码 —— 纯源码解析博客

服务down机了，线程池中的数据如何保证不丢失？

前言