需求分析：

在javashop電商系統(tǒng)中，各種促銷活動(dòng)都有開(kāi)始時(shí)間和結(jié)束時(shí)間,想要讓一個(gè)活動(dòng)在預(yù)定的時(shí)間開(kāi)始或結(jié)束,使用定時(shí)任務(wù)輪詢,存在耗性能并且不能在準(zhǔn)確的時(shí)間點(diǎn)開(kāi)始或結(jié)束的缺點(diǎn),為了可以在指定的時(shí)間執(zhí)行,要求使用延時(shí)任務(wù)

思路：

延時(shí)任務(wù)：指定某日期執(zhí)行某自定義任務(wù)

思路為采用Rabbitmq中的死信轉(zhuǎn)移隊(duì)列的技術(shù)點(diǎn)實(shí)現(xiàn)。

第一步向一個(gè)隊(duì)列(具有xxxx屬性)發(fā)送消息，這個(gè)隊(duì)列的消息可以指定失效時(shí)間

當(dāng)失效發(fā)生時(shí)rabbbitmq會(huì)將此消息轉(zhuǎn)移到另外的一個(gè)普通對(duì)列中，此時(shí)立刻被消費(fèi)了，以此實(shí)現(xiàn)任務(wù)的延遲執(zhí)行。

AMQP 延時(shí)任務(wù)核心類圖

TimeTrigger 觸發(fā)器接口，對(duì)外提供定義延遲任務(wù)的接口，調(diào)用者直接面向此接口。

目前只實(shí)現(xiàn)了基于RabbitMq的實(shí)現(xiàn)，如果有其他延時(shí)任務(wù)實(shí)現(xiàn)(如基于redis)，面向此接口開(kāi)發(fā)即可，定義新增、編輯、刪除任務(wù)操作。

RabbitmqTimeTrigger

基于rabbitmq延時(shí)任務(wù)實(shí)現(xiàn)

TimeTriggerConfig，rabbitmq配置

TimeTriggerMsg，rabbitmq延時(shí)任務(wù)消息

執(zhí)行器類圖

TimeTriggerConsumer 延時(shí)任務(wù)消費(fèi)者，負(fù)責(zé)延時(shí)任務(wù)的調(diào)用

TimeTriggerExecuter 延時(shí)任務(wù)執(zhí)行器接口，自定義延時(shí)任務(wù)需要實(shí)現(xiàn)此接口

PintuanTimeTriggerExecuter 以拼團(tuán)業(yè)務(wù)為例，延時(shí)任務(wù)執(zhí)行的實(shí)現(xiàn)。

新增任務(wù)時(shí)序圖

步驟說(shuō)明：

1、新增延時(shí)任務(wù)，指定延時(shí)任務(wù)所需的參數(shù)(執(zhí)行器beanName，執(zhí)行器參數(shù)，執(zhí)行日期，執(zhí)行任務(wù)標(biāo)識(shí)KEY)

2、rabbitmq發(fā)送消息，將執(zhí)行器以及參數(shù)封裝

3、寫(xiě)入redis，標(biāo)識(shí)任務(wù)需要執(zhí)行

4、mq監(jiān)聽(tīng) 指定時(shí)間任務(wù)

5、消費(fèi)者獲取redis的任務(wù)標(biāo)識(shí)

7、進(jìn)行標(biāo)識(shí)判斷，如果判斷無(wú)效，則不執(zhí)行任務(wù)，return

8、如果任務(wù)標(biāo)識(shí)有效，則通過(guò)springbean容器獲取執(zhí)行器，執(zhí)行execute方法

編輯任務(wù)流程圖

步驟說(shuō)明：

1、編輯延時(shí)任務(wù)，指定延時(shí)任務(wù)所需的參數(shù)(執(zhí)行器，執(zhí)行器參數(shù)，執(zhí)行日期，執(zhí)行任務(wù)標(biāo)識(shí)KEY)

2、刪除redis中的任務(wù)標(biāo)識(shí)，代表任務(wù)取消

3、rabbitmq發(fā)送消息，將執(zhí)行器以及參數(shù)封裝

4、寫(xiě)入redis，標(biāo)識(shí)任務(wù)需要執(zhí)行

5、mq監(jiān)聽(tīng) 指定時(shí)間任務(wù)

7、消費(fèi)者獲取redis的任務(wù)標(biāo)識(shí)

8、進(jìn)行標(biāo)識(shí)判斷，如果判斷無(wú)效，則不執(zhí)行任務(wù)，return

9、如果任務(wù)標(biāo)識(shí)有效，則通過(guò)springbean容器獲取執(zhí)行器，執(zhí)行execute方法

刪除任務(wù)流程圖

步驟說(shuō)明：

1、刪除延時(shí)任務(wù)，參數(shù)(執(zhí)行任務(wù)標(biāo)識(shí)KEY)

2、刪除redis中的任務(wù)標(biāo)識(shí)，代表任務(wù)取消

源碼

TimeTriggerConsumer 延時(shí)任務(wù)消費(fèi)者，負(fù)責(zé)延時(shí)任務(wù)的調(diào)用

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;/** * 延時(shí)任務(wù)生產(chǎn) rabbitmq實(shí)現(xiàn) * @Description: 原理：利用amqp的死信隊(duì)列的超時(shí)屬性，將超時(shí)的任務(wù)轉(zhuǎn)到普通隊(duì)列交給消費(fèi)者執(zhí)行。
* 添加任務(wù)，將任務(wù)執(zhí)行標(biāo)識(shí)、beanid、執(zhí)行時(shí)間，hash值存入redis，標(biāo)識(shí)任務(wù)需要執(zhí)行
* 添加刪除，刪除redis中的任務(wù)標(biāo)識(shí)，消費(fèi)者執(zhí)行時(shí)獲取不到 redis中的標(biāo)識(shí)，則不會(huì)執(zhí)行延時(shí)任務(wù) *

*/@Componentpublic class RabbitmqTimeTrigger implements TimeTrigger { /** * 引入rabbit的操作模板 */ @Autowired private RabbitTemplate rabbitTemplate; @Autowired private Cache cache; private final Logger logger = LoggerFactory.getLogger(getClass()); /** * 添加延時(shí)任務(wù) * * @param executerName 執(zhí)行器 * @param param 執(zhí)行參數(shù) * @param triggerTime 執(zhí)行時(shí)間 * @param uniqueKey 如果是一個(gè) 需要有 修改/取消 延時(shí)任務(wù)功能的延時(shí)任務(wù)，
* 請(qǐng)?zhí)顚?xiě)此參數(shù)，作為后續(xù)刪除，修改做為唯一憑證
* 建議參數(shù)為：PINTUAZN_{ACTIVITY_ID} 例如 pintuan_123
* 業(yè)務(wù)內(nèi)全局唯一 */ @Override public void add(String executerName, Object param, Long triggerTime, String uniqueKey) { if (StringUtil.isEmpty(uniqueKey)) { uniqueKey = StringUtil.getRandStr(10); } //標(biāo)識(shí)任務(wù)需要執(zhí)行 cache.put(RabbitmqTriggerUtil.generate(executerName, triggerTime, uniqueKey), 1); TimeTriggerMsg timeTriggerMsg = new TimeTriggerMsg(executerName, param, triggerTime, uniqueKey); logger.debug("定時(shí)執(zhí)行在【" + DateUtil.toString(triggerTime, "yyyy-MM-dd HH:mm:ss") + "】，消費(fèi)【" + param.toString() + "】"); rabbitTemplate.convertAndSend(TimeTriggerConfig.DELAYED_EXCHANGE_XDELAY, TimeTriggerConfig.DELAY_ROUTING_KEY_XDELAY, timeTriggerMsg, message -> { Long current = DateUtil.getDateline(); //如果執(zhí)行的延時(shí)任務(wù)應(yīng)該是在現(xiàn)在日期之前執(zhí)行的，那么補(bǔ)救一下，要求系統(tǒng)一秒鐘后執(zhí)行 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() + "】執(zhí)行任務(wù)"); return message; }); } /** * 修改延時(shí)任務(wù) * * @param executerName 執(zhí)行器 * @param param 執(zhí)行參數(shù) * @param triggerTime 執(zhí)行時(shí)間 * @param uniqueKey 添加任務(wù)時(shí)的唯一憑證 */ @Override public void edit(String executerName, Object param, Long oldTriggerTime, Long triggerTime, String uniqueKey) { //標(biāo)識(shí)任務(wù)放棄 cache.remove(RabbitmqTriggerUtil.generate(executerName, oldTriggerTime, uniqueKey)); //重新添加任務(wù) this.add(executerName, param, triggerTime, uniqueKey); } /** * 刪除延時(shí)任務(wù) * * @param executerName 執(zhí)行器 * @param triggerTime 執(zhí)行時(shí)間 * @param uniqueKey 添加任務(wù)時(shí)的唯一憑證 */ @Override public void delete(String executerName, Long triggerTime, String uniqueKey) { cache.remove(RabbitmqTriggerUtil.generate(executerName, triggerTime, uniqueKey)); }}

TimeTriggerExecuter 延時(shí)任務(wù)執(zhí)行器接口，自定義延時(shí)任務(wù)需要實(shí)現(xiàn)此接口

package com.enation.app.javashop.framework.trigger.Interface;/** * 延時(shí)任務(wù)執(zhí)行器接口 * */public interface TimeTriggerExecuter { /** * 執(zhí)行任務(wù) * @param object 任務(wù)參數(shù) */ void execute(Object object);}

PintuanTimeTriggerExecuter 以拼團(tuán)業(yè)務(wù)為例，延時(shí)任務(wù)執(zhí)行的實(shí)現(xiàn)。

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;/** * 拼團(tuán)定時(shí)開(kāi)啟關(guān)閉活動(dòng) 延時(shí)任務(wù)執(zhí)行器 * * @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()); /** * 執(zhí)行任務(wù) * * @param object 任務(wù)參數(shù) */ @Override public void execute(Object object) { PintuanChangeMsg pintuanChangeMsg = (PintuanChangeMsg) object; //如果是要開(kāi)啟活動(dòng) 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()); //開(kāi)啟活動(dòng)后，立馬設(shè)置一個(gè)關(guān)閉的流程 pintuanChangeMsg.setOptionType(0); timeTrigger.add(TimeExecute.PINTUAN_EXECUTER, pintuanChangeMsg, pintuan.getEndTime(), "{TIME_TRIGGER}_" + pintuan.getPromotionId()); this.logger.debug("活動(dòng)[" + pintuan.getPromotionName() + "]開(kāi)始,id=[" + pintuan.getPromotionId() + "]"); } } else { //拼團(tuán)活動(dòng)結(jié)束 Pintuan pintuan = pintuanManager.getModel(pintuanChangeMsg.getPintuanId()); if (pintuan.getStatus().equals(PromotionStatusEnum.UNDERWAY.name())) { pintuanManager.closePromotion(pintuanChangeMsg.getPintuanId()); } this.logger.debug("活動(dòng)[" + pintuan.getPromotionName() + "]結(jié)束,id=[" + pintuan.getPromotionId() + "]"); } }} 與50位技術(shù)專家面對(duì)面20年技術(shù)見(jiàn)證，附贈(zèng)技術(shù)全景圖

總結(jié)

以上是生活随笔為你收集整理的机票预定系统类图_电商系统延时任务机制源码分享的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。