spring中自带简单任务调度实现

Z时代
2024-01-10
分类：综合

2. 定时任务调度

@EnableScheduling注解开启对定时任务的支持，使用@Scheduled 注解即可，基于corn、fixedRate、fixedDelay等一些定时策略来实现定时任务

a、fixedDelay控制方法执行的间隔时间，是以上一次方法执行完开始算起，如上一次方法执行阻塞住了，那么直到上一次执行完，并间隔给定的时间后，执行下一次。

b、fixedRate是按照一定的速率执行，是从上一次方法执行开始的时间算起，如果上一次方法阻塞住了，下一次也是不会执行，但是在阻塞这段时间内累计应该执行的次数，当不再阻塞时，一下子把这些全部执行掉，而后再按照固定速率继续执行。

c、cron表达式可以定制化执行任务，类似fixedDelay。也是会按照上一次方法结束时间开始算起。 cron表达式配置：https://www.pppet.net/

@Configuration
@EnableScheduling
public class ScheduleConfig implements SchedulingConfigurer {
    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
	taskRegistrar.setScheduler(threadPoolTaskScheduler());
    }
    @Bean
    public ThreadPoolTaskScheduler threadPoolTaskScheduler() {
	ThreadPoolTaskScheduler executor = new ThreadPoolTaskScheduler();
	executor.setPoolSize(20);
	executor.setThreadNamePrefix("ScheduleExecutor-");
	executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
	// 调度器shutdown被调用时等待当前被调度的任务完成
	executor.setWaitForTasksToCompleteOnShutdown(true);
	// 等待时长
	executor.setAwaitTerminationSeconds(60);
	return executor;
    }
}

@Component
public class NormalScheduledTask {
    /**
     * 类似fixedDelay。也是会按照上一次方法结束时间开始算起。
     */
    @Scheduled(cron = "0/5 * * * * *")
    public void work2() {
        System.out.println("每5秒执行一次" + LocalDateTime.now());
    }
    /**
     * fixedDelay控制方法执行的间隔时间，是以上一次方法执行完开始算起，如上一次方法执行阻塞住了，那么直到上一次执行完，并间隔给定的时间后，执行下一次。
     * 
     */
    @Scheduled(fixedDelay = 1000*3)
    public void work1() {
	 System.out.println("每隔3秒执行一次");
    }
    /**
     * 方法的执行时间超过任务调度频率时，调度器会在当前方法执行完成后立即执行下次任务。
     *  
     */
    @Scheduled(fixedRate = 1000*2)
    public void work3() {
	 System.out.println("每2秒执行一次");
    }
}

动态修改定时配置规则：

@Component
public class DynamicScheduledTask {
    private static final Logger logger = LoggerFactory.getLogger(DynamicScheduledTask.class);
    private final Map<Runnable, ScheduledTask> scheduledTasks = new ConcurrentHashMap<>(16);
    // 利用创建好的调度类统一管理
    @Autowired
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;
    /**
     * 新增定时任务
     * @param task
     * @param cronExpression
     */
    public void addCronTask(Runnable task, String cronExpression) {
        addCronTask(new CronTask(task, cronExpression));
    }
    /**
     * 移除定时任务
     * @param task
     */
    public void removeCronTask(Runnable task) {
        ScheduledTask scheduledTask = this.scheduledTasks.remove(task);
        if (scheduledTask != null)
            scheduledTask.cancel();
    }
    private ScheduledTask scheduleCronTask(CronTask cronTask) {
        ScheduledTask scheduledTask = new ScheduledTask();
        scheduledTask.future = this.threadPoolTaskScheduler.schedule(cronTask.getRunnable(), cronTask.getTrigger());
        return scheduledTask;
    }
    private void addCronTask(CronTask cronTask) {
        if (cronTask != null) {
            Runnable task = cronTask.getRunnable();
            if (this.scheduledTasks.containsKey(task)) {
                removeCronTask(task);
            }
            this.scheduledTasks.put(task, scheduleCronTask(cronTask));
        }
    }
    @PreDestroy
    public void destroy() {
        for (ScheduledTask task : this.scheduledTasks.values()) {
            task.cancel();
        }
        this.scheduledTasks.clear();
    }
}
测试：
   Runnable task = new Runnable() {
	    @Override
	    public void run() {
		System.out.println("执行中" + LocalDateTime.now());
	    }
	};
	dynamicScheduledTask.addCronTask(task, "0/2 * * * * ?");
	Thread.sleep(10000);
	//将task的定时任务改成每3秒进行执行。	dynamicScheduledTask.addCronTask(task, "0/3 * * * * ?");

3.异步任务执行

@EnableAsync来开启异步的支持，使用@Async来对某个方法进行异步执行

@Configuration
@EnableAsync  //实现异步任务
public class ThreadPoolTaskConfig {
	private static final int corePoolSize = 10;       		// 核心线程数（默认线程数）线程池创建时候初始化的线程数
	private static final int maxPoolSize = 100;			    // 最大线程数 线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程
	private static final int keepAliveTime = 10;			// 允许线程空闲时间（单位：默认为秒）当超过了核心线程之外的线程在空闲时间到达之后会被销毁
	private static final int queueCapacity = 200;			// 缓冲队列数 用来缓冲执行任务的队列
	private static final String threadNamePrefix = "Async-Service-"; // 线程池名前缀 方便我们定位处理任务所在的线程池
	@Bean("taskExecutor") // bean的名称，默认为首字母小写的方法名
	public ThreadPoolTaskExecutor taskExecutor(){
		ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
		executor.setCorePoolSize(corePoolSize);   
		executor.setMaxPoolSize(maxPoolSize);
		executor.setQueueCapacity(queueCapacity);
		executor.setKeepAliveSeconds(keepAliveTime);
		executor.setThreadNamePrefix(threadNamePrefix);
		// 线程池对拒绝任务的处理策略 采用了CallerRunsPolicy策略，当线程池没有处理能力的时候，该策略会直接在 execute 方法的调用线程中运行被拒绝的任务；如果执行程序已关闭，则会丢弃该任务
		executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
		// 初始化
		executor.initialize();
		return executor;
	}
}

@Component
@Async
public class AsyncTask {
    //获取异步结果
    public Future<String> task4() throws InterruptedException{
        long begin = System.currentTimeMillis();
        Thread.sleep(2000L);
        long end = System.currentTimeMillis();
        System.out.println("任务4耗时="+(end-begin));
        return new AsyncResult<String>("任务4");
    }
    public Future<String> task5() throws InterruptedException{
        long begin = System.currentTimeMillis();
        Thread.sleep(3000L);
        long end = System.currentTimeMillis();
        System.out.println("任务5耗时="+(end-begin));
        return new AsyncResult<String>("任务5");
    }
    public Future<String> task6() throws InterruptedException{
        long begin = System.currentTimeMillis();
        Thread.sleep(1000L);
        long end = System.currentTimeMillis();
        System.out.println("任务6耗时="+(end-begin));
        return new AsyncResult<String>("任务6");
    }
}