电商系统延时任务机制源码分享

布满荆棘的人生 2022-12-02 15:02 87阅读 0赞

JAVASHOP

# 需求分析： #

在JavaShop电商系统中，各种促销活动都有开始时间和结束时间,想要**让一个活动在预定的时间开始或结束,**使用定时任务轮询,存在耗性能并且不能在准确的时间点开始或结束的缺点,为了可以在指定的时间执行,要求**使用延时任务**

# 思路： #

延时任务：指定某日期执行某自定义任务

**思路为采用Rabbitmq中的死信转移队列的技术点实现。**

第一步向一个队列（具有xxxx属性）发送消息，这个队列的消息可以指定失效时间

当失效发生时rabbbitmq会将此消息转移到另外的一个普通对列中，此时立刻被消费了，以此实现任务的延迟执行。

**AMQP 延时任务核心类图**

![电商系统延时任务机制源码分享][format_png]

TimeTrigger 触发器接口，对外提供定义延迟任务的接口，调用者直接面向此接口。

目前只实现了基于RabbitMq的实现，如果有其他延时任务实现（如基于redis），面向此接口开发即可，定义新增、编辑、删除任务操作。

RabbitmqTimeTrigger

基于rabbitmq延时任务实现

TimeTriggerConfig，rabbitmq配置

TimeTriggerMsg，rabbitmq延时任务消息

**执行器类图**

![电商系统延时任务机制源码分享][format_png 1]

TimeTriggerConsumer 延时任务消费者，负责延时任务的调用

TimeTriggerExecuter 延时任务执行器接口，自定义延时任务需要实现此接口

PintuanTimeTriggerExecuter 以拼团业务为例，延时任务执行的实现。

**新增任务时序图**

![电商系统延时任务机制源码分享][format_png 2]

步骤说明：

1、新增延时任务，指定延时任务所需的参数（执行器beanName，执行器参数，执行日期，执行任务标识KEY）

2、rabbitmq发送消息，将执行器以及参数封装

3、写入redis，标识任务需要执行

4、mq监听 指定时间任务

5、消费者获取redis的任务标识

7、进行标识判断，如果判断无效，则不执行任务，return

8、如果任务标识有效，则通过springbean容器获取执行器，执行execute方法

**编辑任务流程图**

![电商系统延时任务机制源码分享][format_png 3]

步骤说明：

1、编辑延时任务，指定延时任务所需的参数（执行器，执行器参数，执行日期，执行任务标识KEY）

2、删除redis中的任务标识，代表任务取消

3、rabbitmq发送消息，将执行器以及参数封装

4、写入redis，标识任务需要执行

5、mq监听 指定时间任务

7、消费者获取redis的任务标识

8、进行标识判断，如果判断无效，则不执行任务，return

9、如果任务标识有效，则通过springbean容器获取执行器，执行execute方法

**删除任务流程图**

![电商系统延时任务机制源码分享][format_png 4]

步骤说明：

1、删除延时任务，参数（执行任务标识KEY）

2、删除redis中的任务标识，代表任务取消

# 源码 #

TimeTriggerConsumer 延时任务消费者，负责延时任务的调用

package com.enation.app.javashop.framework.trigger.rabbitmq;
 import com.enation.app.javashop.framework.cache.Cache;
 import com.enation.app.javashop.framework.trigger.Interface.TimeTrigger;
 import com.enation.app.javashop.framework.trigger.rabbitmq.model.TimeTriggerMsg;
 import com.enation.app.javashop.framework.trigger.util.RabbitmqTriggerUtil;
 import com.enation.app.javashop.framework.util.DateUtil;
 import com.enation.app.javashop.framework.util.StringUtil;
 import com.enation.app.javashop.framework.logs.Logger;
 import com.enation.app.javashop.framework.logs.LoggerFactory;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 /**
 * 延时任务生产 rabbitmq实现
 * @Description: 原理：利用amqp的死信队列的超时属性，将超时的任务转到普通队列交给消费者执行。 
 * 添加任务，将任务执行标识、beanid、执行时间，hash值存入redis，标识任务需要执行
 * 任务编辑，将之前的标识删除，重新添加任务 
 * 添加删除，删除redis中的任务标识，消费者执行时获取不到 redis中的标识，则不会执行延时任务
 * 
 */
 @Component
 public class RabbitmqTimeTrigger implements TimeTrigger {
 /**
 * 引入rabbit的操作模板
 */
 @Autowired
 private RabbitTemplate rabbitTemplate;
 @Autowired
 private Cache cache;
 private final Logger logger = LoggerFactory.getLogger(getClass());
 /**
 * 添加延时任务
 *
 * @param executerName 执行器
 * @param param 执行参数
 * @param triggerTime 执行时间
 * @param uniqueKey 如果是一个 需要有 修改/取消 延时任务功能的延时任务， 
 * 请填写此参数，作为后续删除，修改做为唯一凭证 
 * 建议参数为：PINTUAZN_{ACTIVITY_ID} 例如 pintuan_123 
 * 业务内全局唯一
 */
 @Override
 public void add(String executerName, Object param, Long triggerTime, String uniqueKey) {
 if (StringUtil.isEmpty(uniqueKey)) {
 uniqueKey = StringUtil.getRandStr(10);
 } //标识任务需要执行
 cache.put(RabbitmqTriggerUtil.generate(executerName, triggerTime, uniqueKey), 1);
 TimeTriggerMsg timeTriggerMsg = new TimeTriggerMsg(executerName, param, triggerTime, uniqueKey);
 logger.debug("定时执行在【" + DateUtil.toString(triggerTime, "yyyy-MM-dd HH:mm:ss") + "】，消费【" + param.toString() + "】");
 rabbitTemplate.convertAndSend(TimeTriggerConfig.DELAYED_EXCHANGE_XDELAY, TimeTriggerConfig.DELAY_ROUTING_KEY_XDELAY, timeTriggerMsg, message -> {
 Long current = DateUtil.getDateline();
 //如果执行的延时任务应该是在现在日期之前执行的，那么补救一下，要求系统一秒钟后执行
 if (triggerTime < current) {
 message.getMessageProperties().setDelay(1000);
 } else {
 Long time = (triggerTime - current) * 1000 + 5000 ;
 message.getMessageProperties().setHeader("x-delay", time);
 }
 logger.debug("还有【" + message.getMessageProperties().getExpiration() + "】执行任务");
 return message;
 });
 }
 /**
 * 修改延时任务
 *
 * @param executerName 执行器
 * @param param 执行参数
 * @param triggerTime 执行时间
 * @param uniqueKey 添加任务时的唯一凭证
 */
 @Override
 public void edit(String executerName, Object param, Long oldTriggerTime, Long triggerTime, String uniqueKey) {
 //标识任务放弃
 cache.remove(RabbitmqTriggerUtil.generate(executerName, oldTriggerTime, uniqueKey));
 //重新添加任务
 this.add(executerName, param, triggerTime, uniqueKey);
 }
 /**
 * 删除延时任务
 *
 * @param executerName 执行器
 * @param triggerTime 执行时间
 * @param uniqueKey 添加任务时的唯一凭证
 */
 @Override
 public void delete(String executerName, Long triggerTime, String uniqueKey) {
 cache.remove(RabbitmqTriggerUtil.generate(executerName, triggerTime, uniqueKey));
 }
 }

TimeTriggerExecuter 延时任务执行器接口，自定义延时任务需要实现此接口

package com.enation.app.javashop.framework.trigger.Interface;
    /**
     * 延时任务执行器接口 *
     */
    public interface TimeTriggerExecuter {
        /**
         * 执行任务
         * @param object 任务参数
         */
        void execute(Object object);
    }

PintuanTimeTriggerExecuter 以拼团业务为例，延时任务执行的实现。

package com.enation.app.javashop.consumer.shop.trigger;
    import com.enation.app.javashop.core.base.message.PintuanChangeMsg;
    import com.enation.app.javashop.core.base.rabbitmq.TimeExecute;
    import com.enation.app.javashop.core.promotion.pintuan.model.Pintuan;
    import com.enation.app.javashop.core.promotion.pintuan.model.PintuanOptionEnum;
    import com.enation.app.javashop.core.promotion.pintuan.service.PintuanManager;
    import com.enation.app.javashop.core.promotion.tool.model.enums.PromotionStatusEnum;
    import com.enation.app.javashop.framework.trigger.Interface.TimeTrigger;
    import com.enation.app.javashop.framework.trigger.Interface.TimeTriggerExecuter;
    import com.enation.app.javashop.framework.logs.Logger;
    import com.enation.app.javashop.framework.logs.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    /**
     * 拼团定时开启关闭活动 延时任务执行器
     *
     * @author Chopper
     * @version v1.0
     * @since v7.0
     * 2019-02-13 下午5:34
     */
    @Component("pintuanTimeTriggerExecute")
    public class PintuanTimeTriggerExecuter implements TimeTriggerExecuter {
        @Autowired
        private TimeTrigger timeTrigger;
        @Autowired
        private PintuanManager pintuanManager;
        private final Logger logger = LoggerFactory.getLogger(getClass());
        /**
         * 执行任务
         *
         * @param object 任务参数
         */
        @Override
        public void execute(Object object) {
            PintuanChangeMsg pintuanChangeMsg = (PintuanChangeMsg) object;        //如果是要开启活动
            if (pintuanChangeMsg.getOptionType() == 1) {
                Pintuan pintuan = pintuanManager.getModel(pintuanChangeMsg.getPintuanId());
                if (PromotionStatusEnum.WAIT.name().equals(pintuan.getStatus()) ||
                        (PromotionStatusEnum.END.name().equals(pintuan.getStatus()) && PintuanOptionEnum.CAN_OPEN.name().equals(pintuan.getOptionStatus()))) {
                    pintuanManager.openPromotion(pintuanChangeMsg.getPintuanId());
                    //开启活动后，立马设置一个关闭的流程
                    pintuanChangeMsg.setOptionType(0);
                    timeTrigger.add(TimeExecute.PINTUAN_EXECUTER, pintuanChangeMsg, pintuan.getEndTime(), "{TIME_TRIGGER}_" + pintuan.getPromotionId());
                    this.logger.debug("活动[" + pintuan.getPromotionName() + "]开始,id=[" + pintuan.getPromotionId() + "]");
                }
            } else {
                //拼团活动结束
                Pintuan pintuan = pintuanManager.getModel(pintuanChangeMsg.getPintuanId());
                if (pintuan.getStatus().equals(PromotionStatusEnum.UNDERWAY.name())) {
                    pintuanManager.closePromotion(pintuanChangeMsg.getPintuanId());
                }
                this.logger.debug("活动[" + pintuan.getPromotionName() + "]结束,id=[" + pintuan.getPromotionId() + "]");
            }
        }
    }

[format_png]: /images/20221123/7477279bd45941fcaa4f17d829d1fa44.png
[format_png 1]: /images/20221123/9669428c3bc041319919ccaffec2b74e.png
[format_png 2]: /images/20221123/1b92e1d8d1a340a2aa14cdbecfd5a0d5.png
[format_png 3]: /images/20221123/663e89b2d6a74e4396ab86880695fe59.png
[format_png 4]: /images/20221123/8f3ae0d917d04425871864870d4fbac5.png