服务重启了，如何保证线程池中的数据不丢失？-数字星空

大家好，我是苏三，又跟大家见面了。

前言

最近有位小伙伴在我的技术群里，问了我一个问题：服务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问题。

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