SpringBoot 使用ThreadPoolExecutor多线程配置和使用

原创
2020/08/21 14:55
阅读数 2.9W

我们常用ThreadPoolExecutor提供的线程池服务,SpringBoot框架提供了@Async注解,帮助我们更方便的将业务逻辑提交到线程池中异步执行,今天我们就来实战体验这个线程池服务。

 

 

一、编写多线程的配置类

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 多线程池配置
 */
@Slf4j
@EnableAsync
@Configuration
public class ThreadExecutorConfig {

    /**
     * SpringBoot会优先使用名称为"taskExecutor"的线程池。
     * 如果没有找到,才会使用其他类型为TaskExecutor或其子类的线程池。
     * @return
     */
    @Bean
    public Executor taskExecutor() {
        log.info("start taskExecutor");

        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 配置核心线程数
        executor.setCorePoolSize(5);
        // 设置最大线程数
        executor.setMaxPoolSize(10);
        // 设置队列容量
        executor.setQueueCapacity(20);
        // 设置线程活跃时间(秒)
        executor.setKeepAliveSeconds(60);
        // 配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix("async-service-");

        // 设置拒绝策略
        // rejection-policy:当pool已经达到max size的时候,如何处理新任务
        // CALLER_RUNS:不在新线程中执行任务,而是有调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 等待所有任务结束后再关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        // 执行初始化
        executor.initialize();

        return executor;
    }

}

java.util.concurrent.TaskExecutor或其子类的bean,并在配置类或直接在程序入口类上声明注解@EnableAsync

 

二、编写Controller请求

import com.piao.sys.sysconfig.service.DemoService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/demo")
public class DemoController {

    @Autowired
    private DemoService demoService;

    @GetMapping(value = "getDemo")
    public void getDemo() {
        demoService.getDemo();
    }

}

 

三、编写Service实现,给服务添加异步注解@Async

public interface DemoService {

    void getDemo();

}
import com.piao.sys.sysconfig.service.DemoService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class DemoServiceImpl implements DemoService {

    @Async
    @Override
    public void getDemo() {
        log.info("this is Thread demo");
    }

}

 

四、请求接口,验证线程

可以看到我们的配置生效了。

实际上@Async还有一个参数,通过Bean名称来指定调用的线程池。比如上例中设置的线程池参数不满足业务需求,可以另外定义合适的线程池,调用时指明使用这个线程池,缺省时springboot会优先使用名称为taskExecutor的线程池,如果没有找到,才会使用其他类型为TaskExecutor或其子类的线程池。

 

五、扩展ThreadPoolTaskExecutor类

虽然我们已经用上了线程池,但是还不清楚线程池当时的情况,有多少线程在执行,多少在队列中等待呢?这里我创建了一个ThreadPoolTaskExecutor的子类,在每次提交线程的时候都会将当前线程池的运行状况打印出来。

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 显示线程执行情况:任务总数、已完成数、活跃线程数,队列大小信息
 */
@Slf4j
public class VisiableThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {

    private void showThreadPoolInfo(String prefix){
        ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor();
        if(null == threadPoolExecutor){
            return;
        }

        log.info("{}{},taskCount [{}], completedTaskCount [{}], activeCount [{}], queueSize [{}]",
                this.getThreadNamePrefix(),
                prefix,
                threadPoolExecutor.getTaskCount(),
                threadPoolExecutor.getCompletedTaskCount(),
                threadPoolExecutor.getActiveCount(),
                threadPoolExecutor.getQueue().size());
    }

    @Override
    public void execute(Runnable task) {
        showThreadPoolInfo("1. do execute");
        super.execute(task);
    }

    @Override
    public void execute(Runnable task, long startTimeout) {
        showThreadPoolInfo("2. do execute");
        super.execute(task, startTimeout);
    }

    @Override
    public Future<?> submit(Runnable task) {
        showThreadPoolInfo("1. do submit");
        return super.submit(task);
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        showThreadPoolInfo("2. do submit");
        return super.submit(task);
    }

    @Override
    public ListenableFuture<?> submitListenable(Runnable task) {
        showThreadPoolInfo("1. do submitListenable");
        return super.submitListenable(task);
    }

    @Override
    public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
        showThreadPoolInfo("2. do submitListenable");
        return super.submitListenable(task);
    }

}
showThreadPoolInfo方法中将任务总数、已完成数、活跃线程数,队列大小都打印出来了。重写了父类的execute、submit等方法。在里面调用showThreadPoolInfo方法,这样每次有任务被提交到线程池的时候,都会将当前线程池的基本情况打印到日志中。

这个时候我们还要修改ThreadExecutorConfig类里面ThreadPoolTaskExecutor。将ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor()改为ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor()。
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 多线程池配置
 */
@Slf4j
@EnableAsync
@Configuration
public class ThreadExecutorConfig {

    /**
     * SpringBoot会优先使用名称为"taskExecutor"的线程池。
     * 如果没有找到,才会使用其他类型为TaskExecutor或其子类的线程池。
     * @return
     */
    @Bean
    public Executor taskExecutor() {
        log.info("start taskExecutor");

        ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();
        // 配置核心线程数
        executor.setCorePoolSize(5);
        // 设置最大线程数
        executor.setMaxPoolSize(10);
        // 设置队列容量
        executor.setQueueCapacity(20);
        // 设置线程活跃时间(秒)
        executor.setKeepAliveSeconds(60);
        // 配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix("async-service-");

        // 设置拒绝策略
        // rejection-policy:当pool已经达到max size的时候,如何处理新任务
        // CALLER_RUNS:不在新线程中执行任务,而是有调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 等待所有任务结束后再关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        // 执行初始化
        executor.initialize();

        return executor;
    }

}

重新启动项目,我们可以看到如下图:

async-service-2. do submit,taskCount [4], completedTaskCount [4], activeCount [0], queueSize [0]

这说明提交任务到线程池的时候,调用的是submit(Callable task)这个方法。当前已经提交了4个任务,完成了4个,当前有0个线程在处理任务,还剩0个任务在队列中等待,线程池的基本情况一路了然。

展开阅读全文
加载中
点击引领话题📣 发布并加入讨论🔥
打赏
0 评论
1 收藏
1
分享
返回顶部
顶部