快捷導航

springboot定時任務(wù)SchedulingConfigurer異步多線程實現(xiàn)方式

更新時間：2024年04月15日 09:24:33 作者：wfreefish

這篇文章主要介紹了springboot定時任務(wù)SchedulingConfigurer異步多線程實現(xiàn)方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

1、設(shè)計定時任務(wù)數(shù)據(jù)庫(mysql)
2、編寫批量獲取定時任務(wù)配置查詢服務(wù)SysScheduledService
3、在啟動容器中添加@EnableScheduling注解
4、增加調(diào)度任務(wù)配置類實現(xiàn)
5、當數(shù)據(jù)庫配置發(fā)生改變后直接調(diào)用
總結(jié)

1、設(shè)計定時任務(wù)數(shù)據(jù)庫(mysql)

CREATE TABLE `sys_scheduled`  (
  `id` varchar(32) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NOT NULL COMMENT '主鍵',
  `job_name` varchar(200) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '定時任務(wù)名稱',
  `class_name` varchar(200) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '類名',
  `method` varchar(250) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '方法名',
  `cron` varchar(200) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '定時任務(wù)表達式',
  `start_flag` tinyint(1) NULL DEFAULT NULL COMMENT '啟用標記 1啟用 0停用',
  `create_date` varchar(32) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '創(chuàng)建日期',
  `create_by` varchar(32) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '創(chuàng)建人',
  `update_by` varchar(32) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '更新人',
  `update_date` varchar(32) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '更新日期',
  `del_flag` varchar(32) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '刪除標記',
  `temp` varchar(2500) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci NULL DEFAULT NULL COMMENT '備注',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = MyISAM AUTO_INCREMENT = 1 CHARACTER SET = utf8mb3 COLLATE = utf8mb3_general_ci COMMENT = '定時任務(wù)配置表' ROW_FORMAT = Dynamic;

2、編寫批量獲取定時任務(wù)配置查詢服務(wù)SysScheduledService

package com.jdsoft.springbootmybatisplusgenrator.service.crm.impl;
 
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.jdsoft.springbootmybatisplusgenrator.bean.scheduled.SysScheduled;
import com.jdsoft.springbootmybatisplusgenrator.dao.crm.ScheduledMapper;
import com.jdsoft.springbootmybatisplusgenrator.service.crm.SysScheduledService;
import org.springframework.stereotype.Service;
 
 
import java.util.List;
 
/**
 * @author: wxf
 * Date: 2023/8/2
 * Time: 17:16
 * Description:
 */
@Service
public class SysScheduledServiceImpl extends ServiceImpl<ScheduledMapper,SysScheduled> implements SysScheduledService {
    @Override
    public List<SysScheduled> getAllConfig() {
        return this.getBaseMapper().getAllConfig();
    }
 
    @Override
    public SysScheduled getById(String id) {
        LambdaQueryWrapper<SysScheduled> lw = new LambdaQueryWrapper<>();
        lw.eq(SysScheduled::getId,id);
        return this.getBaseMapper().selectOne(lw);
    }
}

2.1 po SysScheduled

package com.jdsoft.springbootmybatisplusgenrator.bean.scheduled;
 
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import lombok.Data;
import java.io.Serializable;
 
@Data
public class SysScheduled  implements Serializable {
 
    @TableId(value = "id",type = IdType.ID_WORKER_STR)
    private String id;		// 主鍵
    @TableId(value = "job_name")  //表里的字段名稱映射到實體類
    private String jobName;		// 定時任務(wù)名稱
    @TableId(value = "class_name")
    private String className;		// 類名
    @TableId(value = "method")
    private String method;		// 方法名
    @TableId(value = "cron")
    private String cron;		// 定時任務(wù)表達式
    @TableId(value = "start_flag")
    private Boolean startFlag;		// 啟用標記
    @TableId(value = "temp")
    private String temp;//備注
}

2.2 mybatisPlus業(yè)務(wù)mapper ScheduledMapper

package com.jdsoft.springbootmybatisplusgenrator.dao.crm;
 
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.jdsoft.springbootmybatisplusgenrator.bean.scheduled.SysScheduled;
import org.apache.ibatis.annotations.Select;
 
import java.util.List;
 
/**
 * @author: wxf
 * Date: 2023/8/2
 * Time: 17:14
 * Description: 定時任務(wù)配置 mapper
 */
public interface ScheduledMapper extends BaseMapper<SysScheduled> {
    @Select("SELECT* FROM sys_scheduled u ")
    List<SysScheduled> getAllConfig();
}

3、在啟動容器中添加@EnableScheduling注解

4、增加調(diào)度任務(wù)配置類實現(xiàn)

SchedulingConfigurer, AsyncConfigurer接口

說明：springboot項目啟動會執(zhí)行configureTasks實現(xiàn)方法代碼

package com.scheduled;
 
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledFuture;
 
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.druid.util.StringUtils;
import com.jdsoft.springbootmybatisplusgenrator.bean.scheduled.SysScheduled;
import com.jdsoft.springbootmybatisplusgenrator.service.crm.SysScheduledService;
import com.wxf.springdemo.redis.RedisUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.config.CronTask;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.stereotype.Component;
import org.springframework.util.ReflectionUtils;
 
/**
 * @Package: com.scheduled
 * @ClassName: ScheduleSetting
 * @Author: wxf
 * @Description: 任務(wù)調(diào)度
 * @Date: 2023/8/1 20:11
 * @Version: 1.0
 */
@Component
public class ScheduleSetting implements SchedulingConfigurer, AsyncConfigurer {
 
    @Autowired
    private SysScheduledService sysScheduledService;
 
    private volatile ScheduledTaskRegistrar scheduledTaskRegistrar;
    //調(diào)度器map
    private Map<String, ScheduledFuture<?>> scheduledFutures = new HashMap<String, ScheduledFuture<?>>();
    // 執(zhí)行任務(wù)map
    private Map<String, CronTask> cronTasks = new HashMap<String, CronTask>();
 
 
    @Autowired
    private RedisUtil redisUtil;
 
    private static List<SysScheduled> scheduleList;
 
    @Override
    public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        //異步執(zhí)行定時任務(wù)
        TaskScheduler taskScheduler = taskScheduler();
        scheduledTaskRegistrar.setTaskScheduler(taskScheduler);
        this.scheduledTaskRegistrar = scheduledTaskRegistrar;
        scheduleList = sysScheduledService.getAllConfig();
        redisUtil.setSysScheduledList("scheduleList", scheduleList);
        refresh(scheduleList);
    }
 
    /**
     * 刷新任務(wù)列表
     *
     * @param scheduleList 待執(zhí)行的任務(wù)列表
     */
    public void refresh(List<SysScheduled> scheduleList) {
        //取消停用的定時策略
        Set<String> keyIds = scheduledFutures.keySet();
        CopyOnWriteArrayList<String> copyKeyIds = CollectionUtil.newCopyOnWriteArrayList(keyIds);
 
        for (String scheduledFutureId : copyKeyIds) {
            if (!taskExit(scheduleList, scheduledFutureId)) {
                //cancel 參數(shù)  false 表示如果調(diào)度器當前執(zhí)行任務(wù)時 不進行打斷    true表示直接打斷進程取消策略
                scheduledFutures.get(scheduledFutureId).cancel(false);
                scheduledFutures.remove(scheduledFutureId);
                cronTasks.remove(scheduledFutureId);
            }
        }
        if (CollectionUtil.isNotEmpty(scheduleList)) {
            for (SysScheduled sysScheduled : scheduleList) {
                //定時corn表達式
                String cron = sysScheduled.getCron();
                String keyId = sysScheduled.getId();
                //任務(wù)狀態(tài) 在新增任務(wù)時進行判斷
                Boolean startFlag = sysScheduled.getStartFlag();
                // 如果啟動狀態(tài)是false的話 , taskExit已進行處理,不需要再進行操作，直接跳下一次循環(huán)
                if (!startFlag) {
                    continue;
                }
                if (StringUtils.isEmpty(cron)) {
                    continue;
                }
                //定時任務(wù)存在的話，并且沒發(fā)生改變的話不進行處理
                if (scheduledFutures.containsKey(keyId) &&
                        cronTasks.get(keyId).getExpression().equalsIgnoreCase(cron)) {
                    continue;
                } else if (scheduledFutures.containsKey(keyId)) {
                    //任務(wù)調(diào)度時間發(fā)生改變，取消當前策略的任務(wù)
                    scheduledFutures.get(keyId).cancel(false);
                    scheduledFutures.remove(keyId);
                    cronTasks.remove(keyId);
                }
                //新建定時任務(wù)
                CronTask task = new CronTask(getRunnable(sysScheduled), cron);
                //將定時任務(wù)分配到對應(yīng)的調(diào)度器
                ScheduledFuture<?> scheduledFuture = scheduledTaskRegistrar.getScheduler().
                        schedule(task.getRunnable(), task.getTrigger());
                //保存緩存任務(wù)map
                cronTasks.put(keyId, task);
                //保存緩存調(diào)度器 map
                scheduledFutures.put(keyId, scheduledFuture);
            }
        }
 
 
    }
 
    /**
     * 判斷調(diào)度器緩存map中是否存在查詢到的配置調(diào)度任務(wù)
     *
     * @param scheduleList 數(shù)據(jù)庫查詢的調(diào)度任務(wù)
     * @param taskId       主鍵id
     * @return 判斷結(jié)果
     */
    private boolean taskExit(List<SysScheduled> scheduleList, String taskId) {
        if (CollectionUtil.isEmpty(scheduleList) || StringUtils.isEmpty(taskId)) {
            return false;
        }
        for (SysScheduled sysScheduled : scheduleList) {
            //id存在 且任務(wù)狀態(tài)是開啟狀態(tài) 判斷任務(wù)存在
            if (taskId.equalsIgnoreCase(sysScheduled.getId()) && sysScheduled.getStartFlag()) {
                return true;
            }
        }
        return false;
    }
 
    /**
     * 設(shè)置異步線程池 任務(wù)調(diào)度器
     *
     * @return 返回任務(wù)調(diào)度器對象
     */
    public ThreadPoolTaskScheduler taskScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.initialize();
        //設(shè)置線程池數(shù)量
        scheduler.setPoolSize(8);
        scheduler.setThreadNamePrefix("Mq-Listeners");
        //等待終止時間
        scheduler.setAwaitTerminationSeconds(60);
        //等待任務(wù)完成關(guān)閉
        scheduler.setWaitForTasksToCompleteOnShutdown(true);
        return scheduler;
    }
 
 
    /**
     * 轉(zhuǎn)換首字母小寫
     *
     * @param str
     * @return
     */
    public static String lowerFirstCapse(String str) {
        char[] chars = str.toCharArray();
        chars[0] += 32;
        return String.valueOf(chars);
    }
 
    /**
     * runnable
     *
     * @param scheduleConfig
     * @return
     */
    private Runnable getRunnable(final SysScheduled scheduleConfig) {
        return new Runnable() {
            @Override
            public void run() {
                Class<?> clazz;
                try {
                    //根據(jù)id查詢對應(yīng)的 啟動狀態(tài)
                    clazz = Class.forName(scheduleConfig.getClassName());
                    String className = lowerFirstCapse(clazz.getSimpleName());
                    Object bean = (Object) ApplicationContextHelper.getBean(className);
                    Method method = ReflectionUtils.findMethod(bean.getClass(), scheduleConfig.getMethod());
                    ReflectionUtils.invokeMethod(method, bean);
                } catch (ClassNotFoundException e) {
                    e.printStackTrace();
                }
            }
        };
    }
}