您的位置:首页 > 编程语言 > Java开发

SpringBoot线程池

2022-05-19 22:27 591 查看 https://www.cnblogs.com/xiegon

1、遇到的场景

  • 提高一下插入表的性能优化,两张表,先插旧的表,紧接着插新的表,若是一万多条数据就有点慢了


2、使用步骤

  • 用Spring提供的对
    ThreadPoolExecutor
    封装的线程池
    ThreadPoolTaskExecutor
    ,直接使用注解启用

配置

@Configuration
@EnableAsync
public class ExecutorConfig {

private static final Logger logger = LoggerFactory.getLogger(ExecutorConfig.class);

@Value("${async.executor.thread.core_pool_size}")
private int corePoolSize;
@Value("${async.executor.thread.max_pool_size}")
private int maxPoolSize;
@Value("${async.executor.thread.queue_capacity}")
private int queueCapacity;
@Value("${async.executor.thread.name.prefix}")
private String namePrefix;

@Bean(name = "asyncServiceExecutor")
public Executor asyncServiceExecutor() {
logger.info("start asyncServiceExecutor");
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
// 配置核心线程数
executor.setCorePoolSize(corePoolSize);
// 配置最大线程数
executor.setMaxPoolSize(maxPoolSize);
// 配置队列大小
executor.setQueueCapacity(queueCapacity);
// 配置线程池中的线程的名称前缀
executor.setThreadNamePrefix(namePrefix);

// rejection-policy:当pool已经达到max size的时候,如何处理新任务
// CALLER_RUNS:不在新线程中执行任务,而是有调用者所在的线程来执行
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
//执行初始化
executor.initialize();
return executor;
}
}

  • @Value
    取值配置是在
    application.properties
    中的
# 异步线程配置
# 配置核心线程数
async.executor.thread.core_pool_size = 5
# 配置最大线程数
async.executor.thread.max_pool_size = 5
# 配置队列大小
async.executor.thread.queue_capacity = 99999
# 配置线程池中的线程的名称前缀
async.executor.thread.name.prefix = async-service-


Demo测试

  • Service接口
public interface AsyncService {

/**
* 执行异步任务
* 可以根据需求,自己加参数拟定
*/
void executeAsync();
}

  • Service实现类
@Service
public class AsyncServiceImpl implements AsyncService {

private static final Logger logger = LoggerFactory.getLogger(AsyncServiceImpl.class);

@Override
@Async("asyncServiceExecutor")
public void executeAsync() {
logger.info("start executeAsync");

System.out.println("异步线程要做的事情");
System.out.println("可以在这里执行批量插入等耗时的事情");

logger.info("end executeAsync");
}
}

  • 在Controller层注入刚刚的Service即可
@Autowired
private AsyncService asyncService;

@GetMapping("/async")
public void async(){
asyncService.executeAsync();
}
  • 使用测试工具测试即可看到相应的打印结果


3、摸索一下

  • 弄清楚线程池当时的情况,有多少线程在执行,多少在队列中等待?
  • 创建一个
    ThreadPoolTaskExecutor
    的子类,在每次提交线程的时候都将当前线程池的运行状况打印出来
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
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;

public class VisiableThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {

private static final Logger logger = LoggerFactory.getLogger(VisiableThreadPoolTaskExecutor.class);

private void showThreadPoolInfo(String prefix) {
ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor();

if (null == threadPoolExecutor) {
return;
}

logger.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
    方法中将任务总数、已完成数、活跃线程数,队列大小都打印出来了,然后Override了父类的execute、submit等方法,在里面调用showThreadPoolInfo方法,这样每次有任务被提交到线程池的时候,都会将当前线程池的基本情况打印到日志中

  • 现在修改

    ExecutorConfig.java
    asyncServiceExecutor
    方法,将
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor()
    改为
    ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor()

@Bean(name = "asyncServiceExecutor")
public Executor asyncServiceExecutor() {
logger.info("start asyncServiceExecutor");
// 在这里进行修改
ThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();
// 配置核心线程数
executor.setCorePoolSize(corePoolSize);
// 配置最大线程数
executor.setMaxPoolSize(maxPoolSize);
// 配置队列大小
executor.setQueueCapacity(queueCapacity);
// 配置线程池中的线程的名称前缀
executor.setThreadNamePrefix(namePrefix);

// rejection-policy:当pool已经达到max size的时候,如何处理新任务
// CALLER_RUNS:不在新线程中执行任务,而是有调用者所在的线程来执行
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
//执行初始化
executor.initialize();
return executor;
}
  • 经最后测试得到的结果:提交任务到线程池的时候,调用的是
    submit(Callable task)
    这个方法,当前已经提交了3个任务,完成了3个,当前有0个线程在处理任务,还剩0个任务在队列中等待
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: