Java 线程池

Posted 2023-03-25 Dazzling

相信在实际工作中，大家对于线程池的使用并不陌生，例如以下几个应用场景：

• 支付成功之后，异步发送短信通知用户；

• 公司的OA系统中，提交某些申请之后，异步发送给各个部门负责人进行审批；

• 请求某个接口时，需要做些日志上报之类的记录。

线程池的使用

下边是一个非常简单的线程池使用案例：

public class ThreadExecutorDemo 

    public static void main(String[] args) 
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                1,
                1,
                1,
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000));
        threadPoolExecutor.submit(new Runnable() 
            @Override
            public void run() 
                System.out.println("this is submitTask");
            
        );
        threadPoolExecutor.execute(new Runnable() 
            @Override
            public void run() 
                System.out.println("this is executeTask");
            
        );
    

通过这段简单的案例代码我们可以知道，线程池常用接口有两类，一种叫 submit，一种叫 execute。这两种方法在执行的时候都不会阻塞到调用线程自身，都能达到异步的效果，但是它们在实现效果方面却有所不同。下边让我们对这两种不同接口做了一些比对，大致如下表所示：

方法名字	返回内容	异常处理情况	是否可以实现异步效果
execute	空	如果出现异常，并且没有加 try-catch 块，线程池会将异常给吞掉。	可以
submit	future对象	如果出现异常，会将异常自动抛出	可以

线程池在实际工作中，可以运用的业务场景非常多，下边我列举了几个非常实用的应用场景和大家分享。

用户注册场景

用户注册之后，当用户信息写入到数据库之后，通过线程池异步发送一条消息通知给到用户身上。

public class UserRegistryThreadExecutorDemo 

    public static final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            1,
            1,
            1,
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000));


    /**
* 用户注册
*/
public static void userRegistry() 
        //插入用户信息
        addUserInfo();
        //异步发送消息通知
        threadPoolExecutor.execute(new Runnable() 
            @Override
            public void run() 
                sendNotifyMsg();
            
        );
        System.out.println("用户注册操作流程结束");
    

    /**
* 模拟信息插入动作
*/
private static void addUserInfo() 
        System.out.println("插入用户信息");
    

    /**
* 发送通知信息
*/
private static void sendNotifyMsg() 
        System.out.println("发送消息通知开始");
        try 
            Thread.sleep(1000);
         catch (InterruptedException e) 
            e.printStackTrace();
        
        System.out.println("发送消息通知结束");
    


    public static void main(String[] args) 
        userRegistry();
    

由于发送消息通知这个操作的过程耗时会比较高，并且其发送的结果不需要百分百返回给到用户那边，那么这个发送的操作就可以交给线程池去处理。

审核场景

在审核场景中，用户点击提交按钮之后，会将审核记录插入到对应的表中，然后通过异步线程发送给到各个部门的负责人。

public class AuditExecutorDemo 

    public static final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            1,
            1,
            1,
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000));


    /**
* 执行审核操作
*/
public static boolean doAudit() 
        createAuditRecord();
        threadPoolExecutor.execute(new Runnable() 
            @Override
            public void run() 
                try 
                    Thread.sleep(1000);
                    System.out.println("发送审核记录给到A部门负责人");
                    Thread.sleep(1000);
                    System.out.println("发送审核记录给到B部门负责人");
                    Thread.sleep(1000);
                    System.out.println("发送审核记录给到C部门负责人");
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
            
        );
        return true;
    

    /**
* 模拟将审核记录写入到数据库中
*/
private static void createAuditRecord() 
        System.out.println("插入审核记录");
    

    public static void main(String[] args) 
        boolean auditStatus = doAudit();
        System.out.println("提交审核记录结果:" + auditStatus);
    

和用户注册用例中比较相似，因为消息发送给到各个负责人的过程是比较耗时的一个操作，所以这一块可以通过线程池异步去完成。

通过这几个实际案例我们可以发现，我们在使用线程池的时候，基本都是将任务往线程池里丢。那么当我们将线程任务丢入到线程池之后，又会发生什么事情呢？下边就需要我们深入了解下线程池的原理了。

线程池原理

这里我用一张动态图来带大家形象地理解线程池的内部执行流程，如下：

其实线程池的核心原理就和生产者消费者差不多，比较相似的点在于，一端不断地提交任务，一端不断地消费任务，而生产者就是我们的调用方，消费者则是线程池内部的worker线程。

整体的执行逻辑大致如下：

• 当生产者提交任务之后，线程会优先分配给空闲的核心线程。

• 如果核心线程都处于忙碌状态，那么新提交的任务就会被放入到一个队列中去。

• 若队列满了之后，再有新的任务提交，则会将新提交的任务放入到非核心线程中处理。

• 假若连非核心线程也都处理不过来的话，那么接下来线程池就会执行拒绝策略。

为了方便大家理解，我整理了一张流程图供大家查看：

在实际使用线程池的时候，我们通常又会使用哪些参数呢？下边我整理了一些供大家参考：

• corePoolSize 核心线程数；

• maximumPoolSize 最大线程数；

• keepAliveTime 线程活跃时间；

• zaResizableCapacityLinkedBlockingQueue 基于 JDK 改造的可伸缩队列；

• allowCoreThreadTimeOut 允许核心线程数超时后被回收；

• preStartCoreThread 是否要在一开始就启动 corePoolSize 数量的线程数

• preStartAllCoreThreads 是否要在一开始就启动 maximumPoolSize 数量的线程数

在结合不同的应用场景下，我们通常会给线程池配置不同的参数，对于线程池参数的配置其实也是一门学问，常见的任务类型可以划分为 IO密集型和 CPU密集型。 通常 IO 密集型任务主要是一些涉及到读写磁盘，网络通信之类的任务，这类任务在互联网业务场景中会比较多，例如远程服务调用。而 CPU 密集型任务的代表如：数据加密解密计算、图片识别等。

关于线程池参数方面的设置，其实网上有许多种说法，有的会根据一套公式来给出定义，但其实线程池的参数应该可以动态调整才是合理的。

不同的应用中间件是如何使用线程池的？

Spring 内部的异步注解

在Spring框架中，有一个叫做 @Async 的注解 ，在 Spring 应用上下文中加入一个 @EnableAsync 的注解，接着在使用的方法头部加入一个 @Async 注解，则被标示的方法在执行的时候，就会被自动提交到一个异步线程池中使用，从而产生异步的效果。例如下边这段代码：

@Service
public class AsyncService 

    //加入注解之后，该方法自动就带有了异步的效果
    @Async
    public void testAsync()
        try 
            Thread.sleep(1000*2);
         catch (InterruptedException e) 
            e.printStackTrace();
        
        System.out.println(Thread.currentThread().getName()+" async test");
    

控制器部分代码如下：

@RestController
@RequestMapping(value = "/test")
public class TestController 

    @Resource
    private AsyncService asyncService;

    @GetMapping(value = "/do-test")
    public String doTest()
        //每次调用都会将该方法扔入一个线程池中执行    
        asyncService.testAsync();
        return "success";
    

在生产环境中，我们通常不会直接使用 Spring 自带的线程池，而是会采用一个自定义的线程池去替代它。这是因为 Spring 对于带有 @Async 注解的方法，会将它们丢入到一个叫做 applicationTaskExecutor 的线程池中。

对于这个线程池的属性，我们可以看出，它的内部对于最大线程数设置是一个巨大的数值，因此使用@Async注解默认的线程池的话，容易产生无止境地工作，从而导致oom异常。

为了解决这一存在的风险，通常我们可以尝试将Async对应的线程池给更换为自定义的一个线程池，从而避免 oom 情况的发生，具体操作如下所示：

@Configuration
public class AsyncExecuteConfig extends AsyncConfigurerSupport 

    @Bean
    public ThreadPoolTaskExecutor asyncExecutor() 
        ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
        threadPool.setCorePoolSize(3);
        threadPool.setMaxPoolSize(3);
        threadPool.setWaitForTasksToCompleteOnShutdown(true);
        threadPool.setAwaitTerminationSeconds(60 * 15);
        return threadPool;
    

    @Override
    public Executor getAsyncExecutor() 
        return asyncExecutor();
    

Dubbo 底层的线程池使用

在 Dubbo 线程池的底层中，其实也存在很多使用线程池场景。在 Dubbo 的内部源代码中，专门对不同类型的线程池做了各种扩展，具体可以通过阅读 Dubbo 的源代码去查看，如下图所示：

Dubbo 底层针对线程池的种类做了一系列的封装，并且定义了一个专门的 SPI 接口去进行扩展。

RocketMQ 的消费线程池

在 RocketMQ 的内部也存在着使用线程池的场景，在 Consumer 节点中，会有一个叫做 consumeExecutor 的线程池负责将消息从 broker 中拉取到本地内存进行消费。

初此之外，在构建对象 ConsumeMessageConcurrentlyService 的时候，就默认构建了一批线程池。

通过阅读不同的中间件我们可以发现，线程池在实际应用中是一款实用性非常高的技术，合理地应用线程池技术，可以很好地帮助我们解决一些复杂的应用场景。

总结：

• Dubbo 底层使用了业务线程池和 IO 线程池进行分工合作，IO 线程池一般负责接收请求并且转发，业务线程池则是用于接收请求并且进行业务处理。

• Spring 底层采用了线程池来异步化操作，但是要注意使用 @Async 注解的时候，其底层是会使用一个无界队列的线程池，所以要注意 oom 异常的发生。

• 在 RocketMQ 中，消费者端定义了一个线程池，用于从 broker 节点上拉取消息数据，具体源代码可以在 ConsumeMessageConcurrentlyService 类中查看到。

在使用线程池的时候有没有遇到过什么问题呢？

• 线程池没有做优雅关闭，导致部分运行到一半的任务被杀掉；

• 线程池执行submit的时候，内部出了异常，但是没有加入try-catch代码块，导致异常被吞；

• 线程池的队列长度配置过大，导致应用服务出现 oom；

• 线程池的内部希望获取到外部的线程变量，可以使用 TransmittableThreadLocal 来获取；

• 高并发场景下不适合使用线程池接收任务，适合使用 MQ 来替代。