code-image-backup-public/lilishop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

延时任务规范，任务调度机制，修改为consumer执行，不再被每个进程监控。

Browse Source
This commit is contained in:
Chopper
2021-06-11 11:59:21 +08:00
parent 9195292aa9
commit ebebbcac69
22 changed files with 271 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
consumer/src/main/java/cn/lili/trigger/AbstractDelayQueueListen.java Normal file
View File

@@ -0,0 +1,94 @@
package cn.lili.trigger;
import cn.hutool.json.JSONUtil;
import cn.lili.common.cache.Cache;
import cn.lili.common.utils.ThreadPoolUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.util.CollectionUtils;
import javax.annotation.PostConstruct;
import java.util.Set;
import java.util.concurrent.TimeUnit;
/**
* 延时队列工厂
*
* @author paulG
* @since 2020/11/7
**/
@Slf4j
public abstract class AbstractDelayQueueListen {
@Autowired
private Cache cache;
/**
* 延时队列机器开始运作
*/
private void startDelayQueueMachine() {
log.info("延时队列机器{}开始运作", setDelayQueueName());
// 监听redis队列
while (true) {
try {
// 获取当前时间的时间戳
long now = System.currentTimeMillis() / 1000;
// 获取当前时间前需要执行的任务列表
Set<DefaultTypedTuple> tuples = cache.zRangeByScore(setDelayQueueName(), 0, now);
// 如果任务不为空
if (!CollectionUtils.isEmpty(tuples)) {
log.info("执行任务:{}", JSONUtil.toJsonStr(tuples));
for (DefaultTypedTuple tuple : tuples) {
String jobId = (String) tuple.getValue();
// 移除缓存，如果移除成功则表示当前线程处理了延时任务，则执行延时任务
Long num = cache.zRemove(setDelayQueueName(), jobId);
// 如果移除成功, 则执行
if (num > 0) {
ThreadPoolUtil.execute(() -> invoke(jobId));
}
}
}
} catch (Exception e) {
log.error("处理延时任务发生异常,异常原因为{}", e.getMessage(), e);
} finally {
// 间隔一秒钟搞一次
try {
TimeUnit.SECONDS.sleep(5L);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
/**
* 最终执行的任务方法
*
* @param jobId 任务id
*/
public abstract void invoke(String jobId);
/**
* 要实现延时队列的名字
*/
public abstract String setDelayQueueName();
/**
* 监听队列
*/
@PostConstruct
public void init() {
new Thread(this::startDelayQueueMachine).start();
}
}

5
consumer/src/main/java/cn/lili/trigger/TimeTriggerConsumer.java
View File

@@ -2,9 +2,8 @@ package cn.lili.trigger;
import cn.hutool.json.JSONUtil;
import cn.lili.common.cache.Cache;
import cn.lili.common.trigger.interfaces.TimeTriggerExecutor;
import cn.lili.common.trigger.model.TimeTriggerMsg;
import cn.lili.common.trigger.util.TimeTriggerUtil;
import cn.lili.common.trigger.util.DelayQueueTools;
import cn.lili.common.utils.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
@@ -28,7 +27,7 @@ public class TimeTriggerConsumer implements RocketMQListener<TimeTriggerMsg> {
@Override
public void onMessage(TimeTriggerMsg timeTriggerMsg) {
try {
String key = TimeTriggerUtil.generateKey(timeTriggerMsg.getTriggerExecutor(), timeTriggerMsg.getTriggerTime(), timeTriggerMsg.getUniqueKey());
String key = DelayQueueTools.generateKey(timeTriggerMsg.getTriggerExecutor(), timeTriggerMsg.getTriggerTime(), timeTriggerMsg.getUniqueKey());
if (cache.get(key) == null) {
log.info("执行器执行被取消：{} | 任务标识：{}", timeTriggerMsg.getTriggerExecutor(), timeTriggerMsg.getUniqueKey());

18
consumer/src/main/java/cn/lili/trigger/TimeTriggerExecutor.java Normal file
View File

@@ -0,0 +1,18 @@
package cn.lili.trigger;
/**
* 延时任务执行器接口
*
* @author Chopper
*/
public interface TimeTriggerExecutor {
/**
* 执行任务
*
* @param object 任务参数
*/
void execute(Object object);
}

9
consumer/src/main/java/cn/lili/trigger/executor/PromotionTimeTriggerExecutor.java
View File

@@ -1,13 +1,12 @@
package cn.lili.trigger.executor;
import cn.hutool.json.JSONUtil;
import cn.lili.common.trigger.interfaces.TimeTrigger;
import cn.lili.trigger.TimeTriggerExecutor;
import cn.lili.common.trigger.message.PintuanOrderMessage;
import cn.lili.common.trigger.message.PromotionMessage;
import cn.lili.common.trigger.interfaces.TimeTrigger;
import cn.lili.common.trigger.interfaces.TimeTriggerExecutor;
import cn.lili.common.trigger.model.TimeExecuteConstant;
import cn.lili.common.trigger.model.TimeTriggerMsg;
import cn.lili.common.utils.DateUtil;
import cn.lili.config.rocketmq.RocketmqCustomProperties;
import cn.lili.modules.order.order.service.OrderService;
import cn.lili.modules.promotion.entity.enums.PromotionStatusEnum;
@@ -58,8 +57,10 @@ public class PromotionTimeTriggerExecutor implements TimeTriggerExecutor {
// 结束时间（延时一分钟）
long closeTime = promotionMessage.getEndTime().getTime() + 60000;
TimeTriggerMsg timeTriggerMsg = new TimeTriggerMsg(TimeExecuteConstant.PROMOTION_EXECUTOR, closeTime, promotionMessage, uniqueKey, rocketmqCustomProperties.getPromotionTopic());
timeTrigger.addDelay(timeTriggerMsg, DateUtil.getDelayTime(promotionMessage.getEndTime().getTime()));
//添加延时任务
timeTrigger.addDelay(timeTriggerMsg);
} else {
//不是开始，则修改活动状态
promotionService.updatePromotionStatus(promotionMessage);
}
return;

34
consumer/src/main/java/cn/lili/trigger/listen/PromotionDelayQueueListen.java Normal file
View File

@@ -0,0 +1,34 @@
package cn.lili.trigger.listen;
import cn.hutool.json.JSONUtil;
import cn.lili.common.trigger.enums.DelayQueueEnums;
import cn.lili.common.trigger.interfaces.TimeTrigger;
import cn.lili.common.trigger.model.TimeTriggerMsg;
import cn.lili.trigger.AbstractDelayQueueListen;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* PromotionTimeTriggerListen
*
* @author Chopper
* @version v1.0
* 2021-06-11 10:47
*/
@Component
public class PromotionDelayQueueListen extends AbstractDelayQueueListen {
@Autowired
private TimeTrigger timeTrigger;
@Override
public void invoke(String jobId) {
timeTrigger.execute(JSONUtil.toBean(jobId, TimeTriggerMsg.class));
}
@Override
public String setDelayQueueName() {
return DelayQueueEnums.PROMOTION_QUEUE.name();
}
}