Java使用定时任务开始-竣事活动

写过一篇  金牌会员 | 2024-7-26 14:13:43 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 504|帖子 504|积分 1512

        有一种场景，好比抢购活动，活动的开始肯定不是到点后手动点击开始按钮的，而是通过配置活动开始时间和竣事时间，然后开启定时任务，通过定时任务来控制活动的开始和竣事。

1. 活动上架--触发定时任务的开启

  1. @Override
  2. @Transactional(rollbackFor = Exception.class)
  3. public void publishIntegralActivity(List<Long> idList) {
  4.     // 修改上架状态
  5.     String accountName = SessionUtils.getAccountNameSession();
  6.     integralActivityDao.updatePublishStatus(idList, PublishStatusEnum.PUBLISH_YES.getCode(), accountName);
  7.     // 活动开始定时任务
  8.     for (Long id : idList) {
  9.         IntegralActivity integralActivity = integralActivityDao.getById(id);
  10.         integralActivityCronService.addStartCron(id, integralActivity.getStartTime());
  11.     }
  12. }
复制代码
2. 开启活动开始定时任务

  1. public void addStartCron(long id, Date startTime) {
  2.     long millSecond = startTime.getTime() - System.currentTimeMillis();
  3.     ScheduledFuture<?> scheduledFuture = scheduleExecutor.schedule(new Runnable() {
  4.         @Override
  5.         public void run() {
  6.             log.debug("integral activity[id=" + id + "] start cron begin execute");
  7.             String key = RedisConstant.INTEGRAL_ACTIVITY_CRON + ":" + ACTIVITY_START_CRON_PREFX + id;
  8.             String value = UUID.randomUUID().toString().replace("-", "");
  9.             if (!getExecuteLock(key, value)) {
  10.                 return;
  11.             }
  12.             try {
  13.                 IntegralActivityQueryDto queryDto = new IntegralActivityQueryDto();
  14.                 queryDto.setId(id);
  15.                 List<IntegralActivity> activityList = integralActivityDao.getActivityList(queryDto);
  16.                 if (CollectionUtils.isEmpty(activityList) ||
  17.                         !ActivityStatusEnum.NOT_START.getCode().equals(activityList.get(0).getActivityStatus()) ||
  18.                         startTime.getTime() != activityList.get(0).getStartTime().getTime()) {
  19.                     log.warn("when execute start cron, integral activity[id=" + id + "] not exists or status not match");
  20.                     lockUtils.unLock(key, value);
  21.                     return;
  22.                 }
  23.                 // 更改活动状态
  24.                 integralActivityDao.updateActivityStatus(Lists.newArrayList(id), ActivityStatusEnum.IN_PROGRESS.getCode(), null, null, activityList.get(0).getEndTime());
  25.                 // 清除活动结束定时任务
  26.                 clearCronById(ACTIVITY_END_CRON_PREFX + id);
  27.                 // 添加活动结束定时任务
  28.                 addEndCron(id, activityList.get(0).getEndTime());
  29.                 lockUtils.unLock(key, value);
  30.             } catch (Exception e) {
  31.                 log.error("integral activity[id=" + id + "] start error, retry", e);
  32.                 lockUtils.unLock(key, value);
  33.                 addStartCron(id, startTime);
  34.             }
  35.         }
  36.     }, millSecond, TimeUnit.MILLISECONDS);
  37.     // 加入任务
  38.     currentCronMap.put(ACTIVITY_START_CRON_PREFX + id, scheduledFuture);
  39. }
复制代码
3. 开启定时任务时要加锁，防止并发哀求

  1. private boolean getExecuteLock(String key, String value) {
  2.     for (int i = 1; i <= 5; i++) {
  3.         try {
  4.             if (lockUtils.getLock(key, value, 10 * 60 * 1000)) {
  5.                 return true;
  6.             }
  7.         } catch (Exception e) {
  8.             log.warn("redis lock[key=" + key + "] get error, continue");
  9.         }
  10.     }
  11.     return false;
  12. }
复制代码
4. 开启活动竣事定时任务

  1. private void addEndCron(long id, Date endTime) {
  2.     long millSecond = endTime.getTime() - System.currentTimeMillis();
  3.     log.debug("integral activity[id=" + id + "] end cron begin execute");
  4.     ScheduledFuture<?> scheduledFuture = scheduleExecutor.schedule(new Runnable() {
  5.         @Override
  6.         public void run() {
  7.             try {
  8.                 IntegralActivityQueryDto queryDto = new IntegralActivityQueryDto();
  9.                 queryDto.setId(id);
  10.                 List<IntegralActivity> activityList = integralActivityDao.getActivityList(queryDto);
  11.                 if (CollectionUtils.isEmpty(activityList) ||
  12.                         !ActivityStatusEnum.IN_PROGRESS.getCode().equals(activityList.get(0).getActivityStatus()) ||
  13.                         endTime.getTime() != activityList.get(0).getEndTime().getTime()) {
  14.                     return;
  15.                 }
  16.                 // 更改活动状态
  17.                 integralActivityDao.updateActivityStatus(Lists.newArrayList(id), ActivityStatusEnum.FINISHED.getCode(), null, null, new Date());
  18.                 // 更改上架状态
  19.                 integralActivityDao.updatePublishStatus(Lists.newArrayList(id), PublishStatusEnum.PUBLISH_NO.getCode(), null);
  20.             } catch (Exception e) {
  21.                 log.error("integral activity[id=" + id + "] end error, retry", e);
  22.                 addEndCron(id, endTime);
  23.             }
  24.         }
  25.     }, millSecond, TimeUnit.MILLISECONDS);
  26.     // 加入任务
  27.     currentCronMap.put(ACTIVITY_END_CRON_PREFX + id, scheduledFuture);
  28. }
复制代码
5.  活动下架

  1. @Override
  2. @Transactional(rollbackFor = Exception.class)
  3. public void offIntegralActivity(List<Long> idList) {
  4.     String accountName = SessionUtils.getAccountNameSession();
  5.     integralActivityDao.updateActivityStatus(idList, ActivityStatusEnum.EARLY_STOP.getCode(), null, accountName, new Date());
  6.     integralActivityDao.updatePublishStatus(idList, PublishStatusEnum.PUBLISH_NO.getCode(), accountName);
  7.     // 活动结束定时任务
  8.     for (Long id : idList) {
  9.         integralActivityCronService.clearActivityCron(id);
  10.     }
  11. }
复制代码
6. 活动下架后要清除定时任务

  1. public void clearActivityCron(long id) {
  2.     this.clearCronById(ACTIVITY_START_CRON_PREFX + id);
  3.     this.clearCronById(ACTIVITY_END_CRON_PREFX + id);
  4. }
  6. @SuppressWarnings("rawtypes")
  7. private void clearCronById(String cronId) {
  8.     if (currentCronMap.get(cronId) == null) {
  9.         log.debug("cron[id=" + cronId + "] not exists");
  10.     }
  11.     Future future = currentCronMap.get(cronId);
  12.     if (null != future && !future.isDone()) {
  13.         future.cancel(true); // 正在运行是否干扰
  14.     }
  15.     currentCronMap.remove(cronId);
  16. }
复制代码
7. 程序启动时开启全部活动的定时任务

  1. @Component
  2. @Slf4j
  3. public class InitIntegralActivityCron implements CommandLineRunner {
  4.    
  5.     @Autowired
  6.     private LockUtils lockUtils;
  7.     @Autowired
  8.     private IntegralActivityCronService integralActivityCronService;
  10.     @Override
  11.     public void run(String... args) throws Exception {
  12.         // 初始化积分活动定时任务
  13.         String key = RedisConstant.LOAD_INTEGRAL_ACTIVITY_CRON;
  14.         String value = UUID.randomUUID().toString().replace("-", "");
  15.         if (!lockUtils.getLock(key, value, 30 * 60 * 1000)) {
  16.             log.info("integral activity cron load lock get failed, return!!!");
  17.             return;
  18.         }
  19.         try {
  20.             if (!integralActivityCronService.createAllActivityCron()) {
  21.                 lockUtils.unLock(key, value);
  22.                 log.error("partial integral activity cron load error, system will exit");
  23.                 System.exit(0);
  24.             }
  25.         } catch (Exception e) {
  26.             log.error("integral activity cron load error, system will exit", e);
  27.             lockUtils.unLock(key, value);
  28.             System.exit(0);
  29.         }
  30.     }
  31. }
复制代码
  1. @Override
  2. public boolean createAllActivityCron() {
  3.     boolean isAllSuccess = true;// 是否都创建成功
  4.     List<String> activityStatusList = new ArrayList<String>();
  5.     activityStatusList.add(ActivityStatusEnum.NOT_START.getCode());
  6.     activityStatusList.add(ActivityStatusEnum.IN_PROGRESS.getCode());
  7.     IntegralActivityQueryDto queryDto = new IntegralActivityQueryDto();
  8.     queryDto.setActivityStatusList(activityStatusList);
  9.     queryDto.setPublishStatusList(Lists.newArrayList(PublishStatusEnum.PUBLISH_YES.getCode()));
  10.     List<IntegralActivity> activityList = integralActivityDao.getActivityList(queryDto);
  11.     if (CollectionUtils.isNotEmpty(activityList)) {
  12.         for (IntegralActivity integralActivity : activityList) {
  13.             long id = integralActivity.getId();
  14.             try {
  15.                 if (ActivityStatusEnum.NOT_START.getCode().equals(integralActivity.getActivityStatus())) {
  16.                     this.addStartCron(id, integralActivity.getStartTime());
  17.                 } else {
  18.                     this.addEndCron(id, integralActivity.getEndTime());
  19.                 }
  20.             } catch (Exception e) {
  21.                 isAllSuccess = false;
  22.                 log.error("integral activity[id=" + id + "] cron create error", e);
  23.             }
  24.         }
  25.     }
  26.     log.info("create integral activity cron end, activity count =" + activityList.size() + " " + (isAllSuccess ? "all success" : "partial success"));
  27.     return isAllSuccess;
  28. }
复制代码

上面是定时任务的完整流程，下面补充完整代码：
1. 定时任务完整代码

  1. @Service
  2. @Slf4j
  3. public class IntegralActivityCronServiceImpl implements IntegralActivityCronService {
  4.    
  5.     // 执行定时对象池
  6.     private ScheduledExecutorService scheduleExecutor = Executors.newScheduledThreadPool(2);
  8.     // 所有的定时任务
  9.     @SuppressWarnings("rawtypes")
  10.     private static Map<String, Future> currentCronMap = new HashMap<String, Future>();
  11.     /**
  12.      * 活动开始定时任务ID前缀
  13.      */
  14.     private static final String ACTIVITY_START_CRON_PREFX = "ACTIVITY_START:";
  15.     /**
  16.      * 活动结束定时任务ID前缀
  17.      */
  18.     private static final String ACTIVITY_END_CRON_PREFX = "ACTIVITY_END:";
  20.     @Autowired
  21.     private IntegralActivityDao integralActivityDao;
  22.     @Autowired
  23.     private LockUtils lockUtils;
  25.     @Override
  26.     public boolean createAllActivityCron() {
  27.         boolean isAllSuccess = true;// 是否都创建成功
  28.         List<String> activityStatusList = new ArrayList<String>();
  29.         activityStatusList.add(ActivityStatusEnum.NOT_START.getCode());
  30.         activityStatusList.add(ActivityStatusEnum.IN_PROGRESS.getCode());
  31.         IntegralActivityQueryDto queryDto = new IntegralActivityQueryDto();
  32.         queryDto.setActivityStatusList(activityStatusList);
  33.         queryDto.setPublishStatusList(Lists.newArrayList(PublishStatusEnum.PUBLISH_YES.getCode()));
  34.         List<IntegralActivity> activityList = integralActivityDao.getActivityList(queryDto);
  35.         if (CollectionUtils.isNotEmpty(activityList)) {
  36.             for (IntegralActivity integralActivity : activityList) {
  37.                 long id = integralActivity.getId();
  38.                 try {
  39.                     if (ActivityStatusEnum.NOT_START.getCode().equals(integralActivity.getActivityStatus())) {
  40.                         this.addStartCron(id, integralActivity.getStartTime());
  41.                     } else {
  42.                         this.addEndCron(id, integralActivity.getEndTime());
  43.                     }
  44.                 } catch (Exception e) {
  45.                     isAllSuccess = false;
  46.                     log.error("integral activity[id=" + id + "] cron create error", e);
  47.                 }
  48.             }
  49.         }
  50.         log.info("create integral activity cron end, activity count =" + activityList.size() + " " + (isAllSuccess ? "all success" : "partial success"));
  51.         return isAllSuccess;
  52.     }
  54.     /**
  55.      * 添加启动定时任务
  56.      *
  57.      * @param id
  58.      * @param startTime
  59.      */
  60.     public void addStartCron(long id, Date startTime) {
  61.         long millSecond = startTime.getTime() - System.currentTimeMillis();
  62.         ScheduledFuture<?> scheduledFuture = scheduleExecutor.schedule(new Runnable() {
  63.             @Override
  64.             public void run() {
  65.                 log.debug("integral activity[id=" + id + "] start cron begin execute");
  66.                 String key = RedisConstant.INTEGRAL_ACTIVITY_CRON + ":" + ACTIVITY_START_CRON_PREFX + id;
  67.                 String value = UUID.randomUUID().toString().replace("-", "");
  68.                 if (!getExecuteLock(key, value)) {
  69.                     return;
  70.                 }
  71.                 try {
  72.                     IntegralActivityQueryDto queryDto = new IntegralActivityQueryDto();
  73.                     queryDto.setId(id);
  74.                     List<IntegralActivity> activityList = integralActivityDao.getActivityList(queryDto);
  75.                     if (CollectionUtils.isEmpty(activityList) ||
  76.                             !ActivityStatusEnum.NOT_START.getCode().equals(activityList.get(0).getActivityStatus()) ||
  77.                             startTime.getTime() != activityList.get(0).getStartTime().getTime()) {
  78.                         log.warn("when execute start cron, integral activity[id=" + id + "] not exists or status not match");
  79.                         lockUtils.unLock(key, value);
  80.                         return;
  81.                     }
  82.                     // 更改活动状态
  83.                     integralActivityDao.updateActivityStatus(Lists.newArrayList(id), ActivityStatusEnum.IN_PROGRESS.getCode(), null, null, activityList.get(0).getEndTime());
  84.                     // 清除活动结束定时任务
  85.                     clearCronById(ACTIVITY_END_CRON_PREFX + id);
  86.                     // 添加活动结束定时任务
  87.                     addEndCron(id, activityList.get(0).getEndTime());
  88.                     lockUtils.unLock(key, value);
  89.                 } catch (Exception e) {
  90.                     log.error("integral activity[id=" + id + "] start error, retry", e);
  91.                     lockUtils.unLock(key, value);
  92.                     addStartCron(id, startTime);
  93.                 }
  94.             }
  95.         }, millSecond, TimeUnit.MILLISECONDS);
  96.         // 加入任务
  97.         currentCronMap.put(ACTIVITY_START_CRON_PREFX + id, scheduledFuture);
  98.     }
  100.     /**
  101.      * 添加结束定时任务
  102.      *
  103.      * @param id
  104.      * @param endTime
  105.      */
  106.     private void addEndCron(long id, Date endTime) {
  107.         long millSecond = endTime.getTime() - System.currentTimeMillis();
  108.         log.debug("integral activity[id=" + id + "] end cron begin execute");
  109.         ScheduledFuture<?> scheduledFuture = scheduleExecutor.schedule(new Runnable() {
  110.             @Override
  111.             public void run() {
  112.                 try {
  113.                     IntegralActivityQueryDto queryDto = new IntegralActivityQueryDto();
  114.                     queryDto.setId(id);
  115.                     List<IntegralActivity> activityList = integralActivityDao.getActivityList(queryDto);
  116.                     if (CollectionUtils.isEmpty(activityList) ||
  117.                             !ActivityStatusEnum.IN_PROGRESS.getCode().equals(activityList.get(0).getActivityStatus()) ||
  118.                             endTime.getTime() != activityList.get(0).getEndTime().getTime()) {
  119.                         return;
  120.                     }
  121.                     // 更改活动状态
  122.                     integralActivityDao.updateActivityStatus(Lists.newArrayList(id), ActivityStatusEnum.FINISHED.getCode(), null, null, new Date());
  123.                     // 更改上架状态
  124.                     integralActivityDao.updatePublishStatus(Lists.newArrayList(id), PublishStatusEnum.PUBLISH_NO.getCode(), null);
  125.                 } catch (Exception e) {
  126.                     log.error("integral activity[id=" + id + "] end error, retry", e);
  127.                     addEndCron(id, endTime);
  128.                 }
  129.             }
  130.         }, millSecond, TimeUnit.MILLISECONDS);
  131.         // 加入任务
  132.         currentCronMap.put(ACTIVITY_END_CRON_PREFX + id, scheduledFuture);
  133.     }
  135.     /**
  136.      * 清除活动定时任务
  137.      *
  138.      * @param id
  139.      */
  140.     public void clearActivityCron(long id) {
  141.         this.clearCronById(ACTIVITY_START_CRON_PREFX + id);
  142.         this.clearCronById(ACTIVITY_END_CRON_PREFX + id);
  143.     }
  145.     /**
  146.      * 通过id清除任务
  147.      *
  148.      * @param cronId
  149.      */
  150.     @SuppressWarnings("rawtypes")
  151.     private void clearCronById(String cronId) {
  152.         if (currentCronMap.get(cronId) == null) {
  153.             log.debug("cron[id=" + cronId + "] not exists");
  154.         }
  155.         Future future = currentCronMap.get(cronId);
  156.         if (null != future && !future.isDone()) {
  157.             future.cancel(true); // 正在运行是否干扰
  158.         }
  159.         currentCronMap.remove(cronId);
  160.     }
  161.    
  162.     /**
  163.      * 获取活动执行锁
  164.      *
  165.      * @param key
  166.      * @param value
  167.      * @return
  168.      */
  169.     private boolean getExecuteLock(String key, String value) {
  170.         for (int i = 1; i <= 5; i++) {
  171.             try {
  172.                 if (lockUtils.getLock(key, value, 10 * 60 * 1000)) {
  173.                     return true;
  174.                 }
  175.             } catch (Exception e) {
  176.                 log.warn("redis lock[key=" + key + "] get error, continue");
  177.             }
  178.         }
  179.         return false;
  180.     }
  181. }
复制代码
2. Redis工具类

  1. @Component()
  2. public class LockUtils {
  4.     @Resource
  5.     private RedisTemplate<String, Object> redisTemplate;
  7.     public LockUtils() {
  8.     }
  10.     public boolean getLock(String key) {
  11.         return this.redisTemplate.opsForValue().setIfAbsent("lock_key_" + key, "lock_default_value", Duration.ofSeconds(60L));
  12.     }
  14.     public boolean getLock(String key, String value) {
  15.         return this.redisTemplate.opsForValue().setIfAbsent("lock_key_" + key, value, Duration.ofSeconds(60L));
  16.     }
  18.     public boolean unLock(String key) {
  19.         return this.redisTemplate.delete("lock_key_" + key);
  20.     }
  22.     public boolean unLock(String key, String value) {
  23.         if (value == null) {
  24.             return this.unLock(key);
  25.         } else {
  26.             return value.equals(this.redisTemplate.opsForValue().get("lock_key_" + key)) ? this.redisTemplate.delete("lock_key_" + key) : false;
  27.         }
  28.     }
  30.     public boolean getLock(String key, long expireTime) {
  31.         return this.redisTemplate.opsForValue().setIfAbsent("lock_key_" + key, "lock_default_value", expireTime, TimeUnit.MILLISECONDS);
  32.     }
  34.     public boolean getLock(String key, String value, long expireTime) {
  35.         return this.redisTemplate.opsForValue().setIfAbsent("lock_key_" + key, value, expireTime, TimeUnit.MILLISECONDS);
  36.     }
  37. }
复制代码
3. 活动状态枚举类

  1. public enum ActivityStatusEnum {
  3.     NOT_START("1", "未开始"),
  4.     IN_PROGRESS("2", "进行中"),
  5.     FINISHED("3", "已结束"),
  6.     EARLY_STOP("4", "提前结束");
  7.    
  8.     private String code;
  9.     private String value;
  11.     private ActivityStatusEnum(String code, String value) {
  12.         this.code = code;
  13.         this.value = value;
  14.     }
  16.     public String getCode() {
  17.         return code;
  18.     }
  20.     public void setCode(String code) {
  21.         this.code = code;
  22.     }
  24.     public String getValue() {
  25.         return value;
  26.     }
  28.     public void setValue(String value) {
  29.         this.value = value;
  30.     }
  32.     /**
  33.      * 根据code获取value
  34.      * @param code
  35.      * @return
  36.      */
  37.     public static String getValueByCode(String code) {
  38.         if (StringUtils.isEmpty(code)) {
  39.             return "";
  40.         }
  41.         for (ActivityStatusEnum item : ActivityStatusEnum.values()) {
  42.             if (item.getCode().equals(code)) {
  43.                 return item.getValue();
  44.             }
  45.         }
  46.         return "";
  47.     }
  48.    
  49.     /**
  50.      * 根据code获取value
  51.      * @param code
  52.      * @return
  53.      */
  54.     public static String getCodeByValue(String value) {
  55.         if (StringUtils.isEmpty(value)) {
  56.             return "";
  57.         }
  58.         for (ActivityStatusEnum item : ActivityStatusEnum.values()) {
  59.             if (item.getValue().equals(value)) {
  60.                 return item.getCode();
  61.             }
  62.         }
  63.         return "";
  64.     }
  65. }
复制代码
其它业务代码没有参考价值不再贴出。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

写过一篇

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表