Spring Boot中使用异步线程池、调度任务线程池的配置

mac2024-03-14 21

转自：https://www.cnblogs.com/slei212/p/10732260.html、https://segmentfault.com/a/1190000012506685、https://www.iteye.com/blog/zw7534313-2435135

从实现的技术上来分类，Java定时任务目前主要有三种：

Java自带的java.util.Timer类，这个类允许调度一个java.util.TimerTask任务。使用这种方式可以让你的程序按照某一个频度执行，但不能在指定时间运行；而且作业类需要集成java.util.TimerTask，一般用的较少。Quartz，这是一个功能比较强大的的调度器，可以让你的程序在指定时间执行，也可以按照某一个频度执行；使用起来需要继承org.springframework.scheduling.quartz.QuartzJobBean，配置稍显复杂，所以，一般会使用spring集成quartz，稍后会详细介绍；Spring3.0以后自带的task，即：spring schedule，可以将它看成一个轻量级的Quartz，而且使用起来比Quartz简单许多。

有时项目中既需要异步任务, 也需要调度任务, 想把这两个异步线程池分来就需要配置两个线程池。调度任务添加 @Scheduled 注解, 需要异步执行的方法添加 @Async 注解

中间遇到点小问题, 异步任务线程池总是不生效, 而是使用的调度任务线程池, 经过查文档不断尝试解决了. 公司利用 slf4j 的 MDC 做链路跟踪, 所以还需要添加前置操作, 使用 TaskDecorator 实现。

代码如下：

AsyncConfig.java

package com.ecej.esmart.autodispatch.config; import com.alibaba.fastjson.JSON; import lombok.extern.slf4j.Slf4j; import org.slf4j.MDC; import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.task.TaskDecorator; import org.springframework.scheduling.annotation.AsyncConfigurer; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import java.util.Map; import java.util.concurrent.Executor; @Slf4j @EnableAsync @Configuration public class AsyncConfig implements AsyncConfigurer { @Bean @Override public Executor getAsyncExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(10); executor.setMaxPoolSize(10); executor.setQueueCapacity(100); executor.setThreadNamePrefix("async-pool-"); executor.setTaskDecorator(new MdcTaskDecorator()); executor.setWaitForTasksToCompleteOnShutdown(true); executor.initialize(); return executor; } class MdcTaskDecorator implements TaskDecorator { @Override public Runnable decorate(Runnable runnable) { Map<String, String> contextMap = MDC.getCopyOfContextMap(); try { if (contextMap != null) { MDC.setContextMap(contextMap); } runnable.run(); } finally { /** 清理后会导致父线程的上下文清空,进入时会复制父线程的内容进行覆盖,可不清理 */ //MDC.clear(); } return runnable; } } @Override public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { return (throwable, method, params) -> { log.error("异步任务异常：方法：{} 参数：{}", method.getName(), JSON.toJSONString(params)); log.error(throwable.getMessage(), throwable); }; } }

SchedulingConfig.java

package com.ecej.esmart.autodispatch.config; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.SchedulingConfigurer; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.config.ScheduledTaskRegistrar; @Slf4j @Configuration @EnableScheduling public class SchedulingConfig implements SchedulingConfigurer { @Override public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) { scheduledTaskRegistrar.setTaskScheduler(taskScheduler()); } @Bean(destroyMethod = "shutdown") public ThreadPoolTaskScheduler taskScheduler() { ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.setPoolSize(5); scheduler.setThreadNamePrefix("dispatch-"); scheduler.setAwaitTerminationSeconds(600); scheduler.setErrorHandler(throwable -> log.error("调度任务发生异常", throwable)); scheduler.setWaitForTasksToCompleteOnShutdown(true); return scheduler; } }

添加定时任务：

①串行方式：

在启动类上加注解：@EnableScheduling即可实现。任务添加@Scheduled注解：@Scheduled接受两种定时的设置，一种是cornexpression，一种是Rate/Delay表达式

②并行方式: 为了提高任务执行效率，可以采用并行方式执行定时任务，任务之间互不影响，只要实现SchedulingConfigurer接口就可以。

@Configuration public class ScheduledConfig implements SchedulingConfigurer { public void configureTasks(ScheduledTaskRegistrar taskRegistrar) { taskRegistrar.setScheduler(setExecutor()); } @Bean(destroyMethod="shutdown") public Executor setExecutor(){ return Executors.newScheduledThreadPool(5); // 5个线程来处理。 } }

SpringBoot 定时任务@Scheduled及SchedulingConfigurer

注：Spring 中，创建定时任务除了使用@Scheduled 注解外，还可以使用 SchedulingConfigurer。

@Schedule 注解有一个缺点，其定时的时间不能动态的改变，而基于 SchedulingConfigurer 接口的方式可以做到。SchedulingConfigurer 接口可以实现在@Configuration 类上，同时不要忘了，还需要@EnableScheduling 注解的支持。

1.基于注解实现方式

@Component @EnableScheduling //开启定时任务 @EnableAsync //开启多线程 public class TimerJob { private Logger log = Logger.getLogger(TimerJob.class); @Async //异步方法；这些方法将在执行的时候，将会在独立的线程中被执行，调用者无需等待它的完成，即可继续其他的操作。 @Scheduled(cron = "* 0/30 * * * ?") //时间规则 public void matterTasks() { //业务逻辑 } }

2.SchedulingConfigurer实现方式　　

2.1、在启动类必须加上@EnableScheduling //开启定时任务

2.2、实现SchedulingConfigurer并重写configureTasks方法

@Component //实现SchedulingConfigurer并重写configureTasks方法 public class OrderJobThread implements SchedulingConfigurer { private Logger log = LoggerFactory.getLogger(OrderJobThread.class); private String cron = "* 0/1 * * * ?"; //调用set方法可动态设置时间规则 public String getCron() { return cron; } public void setCron(String cron) { this.cron = cron; } private String name = "测试"; //调用set方法可以动态设置日志名 public String getName() { return name; } public void setName(String name) { this.name = name; } @Override public void configureTasks(ScheduledTaskRegistrar taskRegistrar) { taskRegistrar.addCronTask(new Runnable() { @Override public void run() { log.warn(name+" --- > 开始"); //业务逻辑 log.warn(name+" --- > 结束"); } }, cron); //加入时间 } }

2.3、设置项目启动后，初始化定时任务执行时间

有时希望项目在启动的时候加载一些系统参数或者方法，就要用到ApplicationRunner

ApplicationRunner是一个接口，我们需要实现它，并重写run()方法，当项目启动时，run()方法便会自动执行

@Component @Order(value = 1) //value值会从小至大的执行 public class TimmerStartController implements ApplicationRunner { private static Logger logger = LoggerFactory.getLogger(TimmerStartController.class); @Autowired private OrderJobThread orderJobThread； //得到定时任务 @Override public void run(ApplicationArguments args) throws Exception { logger.info("=========== 项目启动后，初始化定时任务执行时间 ============="); orderJobThread.setCron("* 0/5 * * * ?"); //根据需求重新赋值时间规则 orderJobThread.setName("ordersTasks"); //赋值name } }

最新回复(0)