package cn.lili.trigger.interfaces.impl;
|
|
import cn.hutool.json.JSONUtil;
|
import cn.lili.cache.Cache;
|
import cn.lili.rocketmq.RocketmqSendCallbackBuilder;
|
import cn.lili.trigger.delay.queue.PromotionDelayQueue;
|
import cn.lili.trigger.interfaces.TimeTrigger;
|
import cn.lili.trigger.model.TimeTriggerMsg;
|
import cn.lili.trigger.util.DelayQueueTools;
|
import cn.lili.common.utils.DateUtil;
|
import cn.lili.common.utils.StringUtils;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.rocketmq.spring.core.RocketMQTemplate;
|
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.messaging.Message;
|
import org.springframework.messaging.support.MessageBuilder;
|
import org.springframework.stereotype.Component;
|
|
/**
|
* 延时任务实现
|
*
|
* @author paulG
|
* @since 2020/11/5
|
**/
|
@Component
|
@Slf4j
|
public class RocketmqTimerTrigger implements TimeTrigger {
|
@Autowired
|
private RocketMQTemplate rocketMQTemplate;
|
@Autowired
|
private Cache<Integer> cache;
|
@Autowired
|
private PromotionDelayQueue promotionDelayQueue;
|
|
|
@Override
|
public void addDelay(TimeTriggerMsg timeTriggerMsg) {
|
//执行器唯一key
|
String uniqueKey = timeTriggerMsg.getUniqueKey();
|
if (StringUtils.isEmpty(uniqueKey)) {
|
uniqueKey = StringUtils.getRandStr(10);
|
}
|
//执行任务key
|
String generateKey = DelayQueueTools.generateKey(timeTriggerMsg.getTriggerExecutor(), timeTriggerMsg.getTriggerTime(), uniqueKey);
|
this.cache.put(generateKey, 1);
|
//设置延时任务
|
if (Boolean.TRUE.equals(promotionDelayQueue.addJob(JSONUtil.toJsonStr(timeTriggerMsg), timeTriggerMsg.getTriggerTime()))) {
|
log.info("延时任务标识： {}", generateKey);
|
log.info("定时执行在【" + DateUtil.toString(timeTriggerMsg.getTriggerTime(), "yyyy-MM-dd HH:mm:ss") + "】，消费【" + timeTriggerMsg.getParam().toString() + "】");
|
} else {
|
log.error("延时任务添加失败:{}", timeTriggerMsg);
|
}
|
}
|
|
@Override
|
public void execute(TimeTriggerMsg timeTriggerMsg) {
|
this.addExecute(timeTriggerMsg.getTriggerExecutor(),
|
timeTriggerMsg.getParam(),
|
timeTriggerMsg.getTriggerTime(),
|
timeTriggerMsg.getUniqueKey(),
|
timeTriggerMsg.getTopic()
|
);
|
}
|
|
/**
|
* 将任务添加到mq，mq异步队列执行。
|
* <p>
|
* 本系统中redis相当于延时任务吊起机制，而mq才是实际的业务消费，执行任务的存在
|
*
|
* @param executorName 执行器beanId
|
* @param param 执行参数
|
* @param triggerTime 执行时间 时间戳 秒为单位
|
* @param uniqueKey 如果是一个 需要有 修改/取消 延时任务功能的延时任务，<br/>
|
* 请填写此参数，作为后续删除，修改做为唯一凭证 <br/>
|
* 建议参数为：COUPON_{ACTIVITY_ID} 例如 coupon_123<br/>
|
* 业务内全局唯一
|
* @param topic rocketmq topic
|
*/
|
private void addExecute(String executorName, Object param, Long triggerTime, String uniqueKey, String topic) {
|
|
TimeTriggerMsg timeTriggerMsg = new TimeTriggerMsg(executorName, triggerTime, param, uniqueKey, topic);
|
Message<TimeTriggerMsg> message = MessageBuilder.withPayload(timeTriggerMsg).build();
|
log.info("延时任务发送信息：{}", message);
|
this.rocketMQTemplate.asyncSend(topic, message, RocketmqSendCallbackBuilder.commonCallback());
|
}
|
|
@Override
|
public void edit(String executorName, Object param, Long oldTriggerTime, Long triggerTime, String uniqueKey, int delayTime, String topic) {
|
this.delete(executorName, oldTriggerTime, uniqueKey, topic);
|
this.addDelay(new TimeTriggerMsg(executorName, triggerTime, param, uniqueKey, topic));
|
}
|
|
@Override
|
public void delete(String executorName, Long triggerTime, String uniqueKey, String topic) {
|
String generateKey = DelayQueueTools.generateKey(executorName, triggerTime, uniqueKey);
|
log.info("删除延时任务{}", generateKey);
|
this.cache.remove(generateKey);
|
}
|
}
|
