异步任务和定时任务


异步任务和定时任务

异步任务(AOP)

在开发系统的过程中,通常会考虑到系统的性能问题,提升系统性能的一个重要思想就是“串行”改“并行”。说起“并行”自然离不开“异步”。

使用方式(共两步)

第一步: 在springboot主 启动类中假如一个@EnableAsync注解用于开启异步任务, 如下

@EnableAsync // 开启异步任务
@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

}

第二步: 然后在需要异步操作的方法上加@Async注解,

如果此注解应用于类上, 即表明这个类中的所有方法在执行时都会异步执行

当在执行用此注解描述的方法的时候, 会开辟一个新的线程来执行这个方法

如下Service层代码

@Service
@Async
public class UserServiceImpl implements UserService {

    @Resource
    private UserDao userDao;

    // 假设保存操作需要5秒钟时间才能完成
    @Override
    public void update1() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        userDao.updateUser(new User());
    }

}

我们在controller中进行调用

@Controller
public class UserController {

    @Resource
    private UserService userService;

    @RequestMapping("/update")
    @ResponseBody
    public String update () {
        userService.update1();
        return "success";
    }

}

当我们访问/update的url时, 页面会瞬间显示success

如果说service中没有@Async注解, 即没有开启异步任务, 则需要等待5秒后, 页面才能显示success

当然, 我们在实际开发中, 不可能没有保存成功就提示成功, 这是只是为了演示

实际应用场景

例如在我们记录用户的操作日志的时候, 不可能等日志记录完成再实现正常的业务, 所以可以把机制的记录改为异步操作即可

获取异步任务结果

假如需要获取业务层异步方法的执行结果,可参考如下代码设计进行实现:

@Override
@Async
public Future<Integer> update1() {
    System.out.println("更新中...");
    try {
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    int row = userDao.updateUser(new User());
    return new AsyncResult<Integer>(row); // 使用AsyncResult封装数据
}

其中,AsyncResult对象可以对异步方法的执行结果进行封装,假如外界需要异步方法结果时,可以通过Future对象的get方法获取结果。

原理及配置

对于@Async注解默认会基于ThreadPoolTaskExecutor对象获取工作线程,然后调用由@Async描述的方法,让方法运行于一个工作线程,以实现异步操作。但是假如系统中的默认拒绝处理策略,任务执行过程的异常处理不能满足我们自身业务需求的话,我可以对异步线程池进行自定义.(SpringBoot中默认的异步配置可以参考自动配置对象TaskExecutionAutoConfiguration).

spring:
  task:
    execution:
      pool:
        core-size: 10 #核心线程数,当池中线程数没达到core-size时,每来一个请求都创建一个新的线程
        queue-capacity: 256 #队列容量,当核心线程都在忙,再来新的任务,会将任务放到队列
        max-size: 128 #当核心线程都在忙,队列也满了,再来新的任务,此时会创建新的线程,直到达到maxSize
        keep-alive: 60s #(加s为秒, 不加为毫秒)当任务高峰过后,有些线程会空闲下来,这空闲现线程达到一定的时间会被释放。
        allow-core-thread-timeout: false # 是否允许核心线程超时
      thread-name-prefix: service-task- # 线程名称前缀

对于spring框架中线程池配置参数的涵义,可以参考ThreadPoolExecutor对象中的解释。

定时任务

项目开发中经常需要执行一些定时任务,比如需要在每天凌晨的时候,分析一次前一天的日志信息,Spring为我们提供了异步执行任务调度的方式,提供了两个接口。

  • TaskExecutor接口
  • TaskScheduler接口

两个注解:

  • @EnableScheduling 开启定时任务
  • @Scheduled 添加定时任务, 参数为cron表达式

cron表达式:

在线生成cron表达式: http://www.bejson.com/othertools/cron/

字段 允许值 允许的特殊字符
0~59 , - * /
0~59 , - * /
小时 0~23 , - * /
日期 1~31 , - * ? / L W C
月份 1~12 , - * /
星期 1~7或SUN-SAT(1=SUN) , - * ? / L W C
年(可选) 1970~2099 , - * /

特殊字符:

特殊字符 含义
, (逗号) 枚举
-(减号) 区间
*(星号) 任意
/(左斜杠) 步长
?(问好) 日/星期冲突匹配
L(大写L) 最后
W(大写W) 工作日
C(大写C) 和calendar联系后计算过的值
#(井号) 星期, 4#2, 第二个星期三

定时任务案例

例如, 我们需要每5秒记录一下日志

@Service // 1. 注入bean
@EnableScheduling   // 2.开启定时任务
public class LogServiceImpl {

    //3.添加定时任务
    //秒    分    时     日    月    周几
    @Scheduled(cron = "0/5 * * * * ?")
    //或直接指定时间间隔,例如:5秒
    //@Scheduled(fixedRate=5000) // 我们也可以使用这种方式(单位: 毫秒)
    private void configureTasks() {
        // 输入日志
        System.err.println("执行静态定时任务时间: " + LocalDateTime.now());
    }

}

注意, 定时任务的类必须交给spring管理

扩展: 自定义异步池(异步任务)

为了让Spring中的异步池更好的服务于我们的业务,同时也尽量避免OOM,可以自定义线程池优化设计如下:关键代码如下:

package com.cy.pj.common.config
@Slf4j
@Setter
@Configuration
@ConfigurationProperties("async-thread-pool")
public class SpringAsyncConfig implements AsyncConfigurer{
    /**核心线程数*/
	private int corePoolSize=20;
	/**最大线程数*/
	private int maximumPoolSize=1000;
	/**线程空闲时间*/
	private int keepAliveTime=30;
	/**阻塞队列容量*/
	private int queueCapacity=200;
	/**构建线程工厂*/
	private ThreadFactory threadFactory=new ThreadFactory() {
		//CAS算法
		private AtomicInteger at = new AtomicInteger(1000);
		@Override
		public Thread newThread(Runnable r) {
			return new Thread(r, 
"db-async-thread-"+at.getAndIncrement());
		}
	};	
    
	@Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(corePoolSize);
        executor.setMaxPoolSize(maximumPoolSize);
        executor.setKeepAliveSeconds(keepAliveTime);
        executor.setQueueCapacity(queueCapacity);
        executor.setRejectedExecutionHandler((Runnable r, 
 ThreadPoolExecutor exe) -> {
                log.warn("当前任务线程池队列已满.");
        });
        executor.initialize();
        return executor;
    }
 
    @Override
    public AsyncUncaughtExceptionHandler 
        		getAsyncUncaughtExceptionHandler() {
        return new AsyncUncaughtExceptionHandler() {
            @Override
            public void handleUncaughtException(
                	Throwable ex ,
                    Method method , 
                	Object... params) {
                log.error("线程池执行任务发生未知异常.", ex);
            }
        };
    }
}

其中:@ConfigurationProperties("async-thread-pool")的含义是读取application.yml配置文件中以"async-thread-pool"名为前缀的配置信息,并通过所描述类的set方法赋值给对应的属性,在application.yml中连接器池的关键配置如下:

async-thread-pool:
       corePoolSize: 20
       maxPoolSize: 1000
       keepAliveSeconds: 30
       queueCapacity: 1000

后续在业务类中,假如我们使用@Async注解描述业务方法,默认会使用ThreadPoolTaskExecutor池对象中的线程执行异步任务。

打赏

免责声明!

本站转载的文章为个人学习借鉴使用,本站对版权不负任何法律责任。如果侵犯了您的隐私权益,请联系本站邮箱yoyou2525@163.com删除。



 
粤ICP备18138465号  © 2018-2021 CODEPRJ.COM