新增mq消息发送失败重试功能
This commit is contained in:
tianfeng
parent
2bf677ebe3
commit
2a05a4ad00
@ -9,6 +9,7 @@ import org.springframework.context.annotation.ComponentScan;
|
||||
@Slf4j
|
||||
@ComponentScan({
|
||||
"com.red.circle.mq.rocket.business.producer",
|
||||
"com.red.circle.mq.rocket.business.failmessage",
|
||||
"com.red.circle.mq.rocket.business.config",
|
||||
"com.red.circle.mq.rocket.business.service"
|
||||
})
|
||||
|
||||
@ -0,0 +1,60 @@
|
||||
package com.red.circle.mq.rocket.business.failmessage;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
/**
|
||||
* RocketMQ 失败消息详情
|
||||
*/
|
||||
@Data
|
||||
public class RocketMqFailMessageDetail {
|
||||
|
||||
/**
|
||||
* 业务唯一标识
|
||||
*/
|
||||
private String bizNo;
|
||||
|
||||
/**
|
||||
* Topic
|
||||
*/
|
||||
private String topic;
|
||||
|
||||
/**
|
||||
* Tag
|
||||
*/
|
||||
private String tag;
|
||||
|
||||
/**
|
||||
* 消息体
|
||||
*/
|
||||
private String messageBody;
|
||||
|
||||
/**
|
||||
* 延迟投递时间（毫秒时间戳）
|
||||
*/
|
||||
private Long startDeliverTime;
|
||||
|
||||
/**
|
||||
* 重试次数
|
||||
*/
|
||||
private Integer retryCount;
|
||||
|
||||
/**
|
||||
* 状态：0-待重试，1-重试中，2-成功，3-失败
|
||||
*/
|
||||
private Integer status;
|
||||
|
||||
/**
|
||||
* 错误信息
|
||||
*/
|
||||
private String errorMsg;
|
||||
|
||||
/**
|
||||
* 创建时间（毫秒时间戳）
|
||||
*/
|
||||
private Long createTime;
|
||||
|
||||
/**
|
||||
* 下次重试时间（毫秒时间戳）
|
||||
*/
|
||||
private Long nextRetryTime;
|
||||
}
|
||||
@ -0,0 +1,33 @@
|
||||
package com.red.circle.mq.rocket.business.failmessage;
|
||||
|
||||
/**
|
||||
* RocketMQ 失败消息 Redis Key
|
||||
*/
|
||||
public class RocketMqFailMessageKeys {
|
||||
|
||||
private static final String PREFIX = "mq:fail:";
|
||||
|
||||
/**
|
||||
* 消息详情 Hash Key
|
||||
* mq:fail:message:{bizNo}
|
||||
*/
|
||||
public static String getMessageDetailKey(String bizNo) {
|
||||
return PREFIX + "message:" + bizNo;
|
||||
}
|
||||
|
||||
/**
|
||||
* 重试队列 ZSet Key
|
||||
* mq:fail:retry:queue
|
||||
*/
|
||||
public static String getRetryQueueKey() {
|
||||
return PREFIX + "retry:queue";
|
||||
}
|
||||
|
||||
/**
|
||||
* Tag 分组 Set Key
|
||||
* mq:fail:tag:{tag}
|
||||
*/
|
||||
public static String getTagGroupKey(String tag) {
|
||||
return PREFIX + "tag:" + tag;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,26 @@
|
||||
package com.red.circle.mq.rocket.business.failmessage;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* RocketMQ 失败消息查询服务
|
||||
*/
|
||||
public interface RocketMqFailMessageQueryService {
|
||||
|
||||
/**
|
||||
* 根据 Tag 获取待重试的消息
|
||||
*
|
||||
* @param tag 消息标签
|
||||
* @param limit 限制数量
|
||||
* @return 待重试消息列表
|
||||
*/
|
||||
List<RocketMqFailMessageDetail> getPendingMessagesByTag(String tag, int limit);
|
||||
|
||||
/**
|
||||
* 根据业务号获取消息详情
|
||||
*
|
||||
* @param bizNo 业务唯一标识
|
||||
* @return 消息详情
|
||||
*/
|
||||
RocketMqFailMessageDetail getMessageByBizNo(String bizNo);
|
||||
}
|
||||
@ -0,0 +1,44 @@
|
||||
package com.red.circle.mq.rocket.business.failmessage;
|
||||
|
||||
/**
|
||||
* RocketMQ 失败消息重试策略
|
||||
*/
|
||||
public class RocketMqFailMessageRetryStrategy {
|
||||
|
||||
/**
|
||||
* 重试间隔：1分钟、5分钟、30分钟
|
||||
*/
|
||||
private static final long[] RETRY_INTERVALS = {
|
||||
60 * 1000L, // 第1次：1分钟后
|
||||
5 * 60 * 1000L, // 第2次：5分钟后
|
||||
30 * 60 * 1000L // 第3次：30分钟后
|
||||
};
|
||||
|
||||
/**
|
||||
* 最大重试次数
|
||||
*/
|
||||
public static final int MAX_RETRY_COUNT = 3;
|
||||
|
||||
/**
|
||||
* 获取下次重试间隔
|
||||
*
|
||||
* @param retryCount 当前重试次数
|
||||
* @return 重试间隔（毫秒）
|
||||
*/
|
||||
public static long getNextRetryInterval(int retryCount) {
|
||||
if (retryCount >= RETRY_INTERVALS.length) {
|
||||
return RETRY_INTERVALS[RETRY_INTERVALS.length - 1];
|
||||
}
|
||||
return RETRY_INTERVALS[retryCount];
|
||||
}
|
||||
|
||||
/**
|
||||
* 计算下次重试时间
|
||||
*
|
||||
* @param retryCount 当前重试次数
|
||||
* @return 下次重试时间（毫秒时间戳）
|
||||
*/
|
||||
public static long calculateNextRetryTime(int retryCount) {
|
||||
return System.currentTimeMillis() + getNextRetryInterval(retryCount);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,38 @@
|
||||
package com.red.circle.mq.rocket.business.failmessage;
|
||||
|
||||
/**
|
||||
* RocketMQ 失败消息状态更新服务
|
||||
*/
|
||||
public interface RocketMqFailMessageUpdateService {
|
||||
|
||||
/**
|
||||
* 锁定消息进行重试（更新状态为重试中）
|
||||
*
|
||||
* @param bizNo 业务唯一标识
|
||||
* @return 是否锁定成功
|
||||
*/
|
||||
boolean lockForRetry(String bizNo);
|
||||
|
||||
/**
|
||||
* 标记消息重试成功
|
||||
*
|
||||
* @param bizNo 业务唯一标识
|
||||
*/
|
||||
void markSuccess(String bizNo);
|
||||
|
||||
/**
|
||||
* 标记消息重试失败（增加重试次数，计算下次重试时间）
|
||||
*
|
||||
* @param bizNo 业务唯一标识
|
||||
* @param errorMsg 错误信息
|
||||
*/
|
||||
void markFailed(String bizNo, String errorMsg);
|
||||
|
||||
/**
|
||||
* 标记消息最终失败（超过最大重试次数）
|
||||
*
|
||||
* @param bizNo 业务唯一标识
|
||||
* @param errorMsg 错误信息
|
||||
*/
|
||||
void markFinalFailed(String bizNo, String errorMsg);
|
||||
}
|
||||
@ -0,0 +1,65 @@
|
||||
package com.red.circle.mq.rocket.business.failmessage.impl;
|
||||
|
||||
import com.red.circle.component.redis.service.RedisService;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageDetail;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageKeys;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageQueryService;
|
||||
import com.red.circle.tool.core.collection.CollectionUtils;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.Set;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* RocketMQ 失败消息查询服务实现
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class RocketMqFailMessageQueryServiceImpl implements RocketMqFailMessageQueryService {
|
||||
|
||||
private final RedisService redisService;
|
||||
|
||||
@Override
|
||||
public List<RocketMqFailMessageDetail> getPendingMessagesByTag(String tag, int limit) {
|
||||
try {
|
||||
// 1. 从 Set 中获取该 tag 的所有 bizNo
|
||||
String tagGroupKey = RocketMqFailMessageKeys.getTagGroupKey(tag);
|
||||
Set<String> bizNos = redisService.setGet(tagGroupKey).stream().map(String::valueOf).collect(Collectors.toSet());
|
||||
|
||||
if (CollectionUtils.isEmpty(bizNos)) {
|
||||
return List.of();
|
||||
}
|
||||
|
||||
long currentTime = System.currentTimeMillis();
|
||||
|
||||
// 2. 批量获取详情并过滤
|
||||
return bizNos.stream()
|
||||
.limit(limit)
|
||||
.map(this::getMessageByBizNo)
|
||||
.filter(Objects::nonNull)
|
||||
.filter(msg -> msg.getStatus() == 0) // 待重试
|
||||
.filter(msg -> msg.getNextRetryTime() != null && msg.getNextRetryTime() <= currentTime)
|
||||
.collect(Collectors.toList());
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("根据Tag查询待重试消息异常，tag={}, error={}", tag, e.getMessage(), e);
|
||||
return List.of();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public RocketMqFailMessageDetail getMessageByBizNo(String bizNo) {
|
||||
try {
|
||||
String messageKey = RocketMqFailMessageKeys.getMessageDetailKey(bizNo);
|
||||
return redisService.hashGet(messageKey, "detail", RocketMqFailMessageDetail.class);
|
||||
} catch (Exception e) {
|
||||
log.error("获取失败消息详情异常，bizNo={}, error={}", bizNo, e.getMessage(), e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,155 @@
|
||||
package com.red.circle.mq.rocket.business.failmessage.impl;
|
||||
|
||||
import com.red.circle.component.redis.service.RedisService;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageDetail;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageKeys;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageRetryStrategy;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageUpdateService;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* RocketMQ 失败消息状态更新服务实现
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class RocketMqFailMessageUpdateServiceImpl implements RocketMqFailMessageUpdateService {
|
||||
|
||||
private final RedisService redisService;
|
||||
|
||||
private static final long EXPIRE_DAYS = 7L;
|
||||
|
||||
@Override
|
||||
public boolean lockForRetry(String bizNo) {
|
||||
try {
|
||||
String messageKey = RocketMqFailMessageKeys.getMessageDetailKey(bizNo);
|
||||
RocketMqFailMessageDetail detail = redisService.hashGet(
|
||||
messageKey, "detail", RocketMqFailMessageDetail.class);
|
||||
|
||||
if (detail == null) {
|
||||
log.warn("消息不存在，无法加锁，bizNo={}", bizNo);
|
||||
return false;
|
||||
}
|
||||
|
||||
// 乐观锁：只有状态为待重试时才能锁定
|
||||
if (detail.getStatus() != 0) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// 更新状态为重试中
|
||||
detail.setStatus(1);
|
||||
redisService.hashPut(messageKey, "detail", detail, EXPIRE_DAYS, TimeUnit.DAYS);
|
||||
|
||||
return true;
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("锁定重试消息异常，bizNo={}, error={}", bizNo, e.getMessage(), e);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void markSuccess(String bizNo) {
|
||||
try {
|
||||
String messageKey = RocketMqFailMessageKeys.getMessageDetailKey(bizNo);
|
||||
RocketMqFailMessageDetail detail = redisService.hashGet(
|
||||
messageKey, "detail", RocketMqFailMessageDetail.class);
|
||||
|
||||
if (detail == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
// 更新状态为成功
|
||||
detail.setStatus(2);
|
||||
redisService.hashPut(messageKey, "detail", detail, EXPIRE_DAYS, TimeUnit.DAYS);
|
||||
|
||||
// 从重试队列中移除
|
||||
String retryQueueKey = RocketMqFailMessageKeys.getRetryQueueKey();
|
||||
redisService.zsetRemove(retryQueueKey, bizNo);
|
||||
|
||||
// 从 Tag 分组中移除
|
||||
String tagGroupKey = RocketMqFailMessageKeys.getTagGroupKey(detail.getTag());
|
||||
redisService.setRemove(tagGroupKey, bizNo);
|
||||
|
||||
log.info("MQ失败消息重试成功，bizNo={}, tag={}", bizNo, detail.getTag());
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("标记消息成功异常，bizNo={}, error={}", bizNo, e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void markFailed(String bizNo, String errorMsg) {
|
||||
try {
|
||||
String messageKey = RocketMqFailMessageKeys.getMessageDetailKey(bizNo);
|
||||
RocketMqFailMessageDetail detail = redisService.hashGet(
|
||||
messageKey, "detail", RocketMqFailMessageDetail.class);
|
||||
|
||||
if (detail == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
int newRetryCount = detail.getRetryCount() + 1;
|
||||
|
||||
// 检查是否超过最大重试次数
|
||||
if (newRetryCount >= RocketMqFailMessageRetryStrategy.MAX_RETRY_COUNT) {
|
||||
markFinalFailed(bizNo, "超过最大重试次数: " + errorMsg);
|
||||
return;
|
||||
}
|
||||
|
||||
// 更新重试信息
|
||||
detail.setRetryCount(newRetryCount);
|
||||
detail.setStatus(0); // 待重试
|
||||
detail.setErrorMsg(errorMsg);
|
||||
|
||||
// 计算下次重试时间
|
||||
long nextRetryTime = RocketMqFailMessageRetryStrategy.calculateNextRetryTime(newRetryCount);
|
||||
detail.setNextRetryTime(nextRetryTime);
|
||||
|
||||
// 更新 Redis
|
||||
redisService.hashPut(messageKey, "detail", detail, EXPIRE_DAYS, TimeUnit.DAYS);
|
||||
|
||||
// 更新重试队列中的时间
|
||||
String retryQueueKey = RocketMqFailMessageKeys.getRetryQueueKey();
|
||||
redisService.zsetAdd(retryQueueKey, bizNo, nextRetryTime);
|
||||
|
||||
log.warn("MQ消息重试失败，将重试，bizNo={}, retryCount={}, nextRetryTime={}, error={}",
|
||||
bizNo, newRetryCount, nextRetryTime, errorMsg);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("标记消息失败异常，bizNo={}, error={}", bizNo, e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void markFinalFailed(String bizNo, String errorMsg) {
|
||||
try {
|
||||
String messageKey = RocketMqFailMessageKeys.getMessageDetailKey(bizNo);
|
||||
RocketMqFailMessageDetail detail = redisService.hashGet(
|
||||
messageKey, "detail", RocketMqFailMessageDetail.class);
|
||||
|
||||
if (detail == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
// 更新状态为最终失败
|
||||
detail.setStatus(3);
|
||||
detail.setErrorMsg(errorMsg);
|
||||
redisService.hashPut(messageKey, "detail", detail, EXPIRE_DAYS, TimeUnit.DAYS);
|
||||
|
||||
// 从重试队列中移除
|
||||
String retryQueueKey = RocketMqFailMessageKeys.getRetryQueueKey();
|
||||
redisService.zsetRemove(retryQueueKey, bizNo);
|
||||
|
||||
log.error("MQ消息最终失败，bizNo={}, tag={}, error={}",
|
||||
bizNo, detail.getTag(), errorMsg);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("标记消息最终失败异常，bizNo={}, error={}", bizNo, e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,74 @@
|
||||
package com.red.circle.mq.rocket.business.failmessage.impl;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.red.circle.component.mq.config.RocketMqFailMessage;
|
||||
import com.red.circle.component.mq.config.RocketMqSenderFailService;
|
||||
import com.red.circle.component.redis.service.RedisService;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageDetail;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageKeys;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageRetryStrategy;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* RocketMQ 失败消息服务实现 - Redis 版本
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class RocketMqSenderFailServiceRedisImpl implements RocketMqSenderFailService {
|
||||
|
||||
private final RedisService redisService;
|
||||
|
||||
private static final long EXPIRE_DAYS = 7L;
|
||||
|
||||
@Override
|
||||
public void add(RocketMqFailMessage failMessage) {
|
||||
try {
|
||||
String bizNo = failMessage.getId();
|
||||
String messageKey = RocketMqFailMessageKeys.getMessageDetailKey(bizNo);
|
||||
|
||||
// 防重检查
|
||||
if (redisService.hashKeys(messageKey).contains(messageKey)) {
|
||||
log.warn("MQ失败消息已存在，bizNo={}", bizNo);
|
||||
return;
|
||||
}
|
||||
|
||||
// 构建消息详情
|
||||
RocketMqFailMessageDetail detail = new RocketMqFailMessageDetail();
|
||||
detail.setBizNo(bizNo);
|
||||
detail.setTopic(failMessage.getTopic());
|
||||
detail.setTag(failMessage.getTag());
|
||||
detail.setMessageBody(JSON.toJSONString(failMessage.getBody()));
|
||||
detail.setStartDeliverTime(failMessage.getStartDeliverTime());
|
||||
detail.setRetryCount(0);
|
||||
detail.setStatus(0); // 待重试
|
||||
detail.setErrorMsg(failMessage.getErrorMsg());
|
||||
detail.setCreateTime(System.currentTimeMillis());
|
||||
|
||||
// 计算下次重试时间
|
||||
long nextRetryTime = RocketMqFailMessageRetryStrategy.calculateNextRetryTime(0);
|
||||
detail.setNextRetryTime(nextRetryTime);
|
||||
|
||||
redisService.hashPut(messageKey, "detail", detail, EXPIRE_DAYS, TimeUnit.DAYS);
|
||||
|
||||
String retryQueueKey = RocketMqFailMessageKeys.getRetryQueueKey();
|
||||
redisService.zsetAdd(retryQueueKey, bizNo, nextRetryTime);
|
||||
|
||||
// 3. 按 Tag 分组（Set）
|
||||
String tagGroupKey = RocketMqFailMessageKeys.getTagGroupKey(failMessage.getTag());
|
||||
redisService.setAdd(tagGroupKey, bizNo);
|
||||
redisService.expire(tagGroupKey, EXPIRE_DAYS, TimeUnit.DAYS);
|
||||
|
||||
log.error("MQ消息发送失败已入Redis，bizNo={}, topic={}, tag={}",
|
||||
bizNo, failMessage.getTopic(), failMessage.getTag());
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("保存MQ失败消息到Redis异常，bizNo={}, error={}",
|
||||
failMessage.getId(), e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,154 @@
|
||||
package com.red.circle.other.app.scheduler;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.red.circle.component.mq.MessageEvent;
|
||||
import com.red.circle.component.mq.service.FixedTopic;
|
||||
import com.red.circle.component.redis.annotation.TaskCacheLock;
|
||||
import com.red.circle.mq.business.model.event.gift.GiveAwayGiftBatchEvent;
|
||||
import com.red.circle.mq.business.model.event.gift.OfflineProcessGiftEvent;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageDetail;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageQueryService;
|
||||
import com.red.circle.mq.rocket.business.failmessage.RocketMqFailMessageUpdateService;
|
||||
import com.red.circle.mq.rocket.business.producer.GiftMqMessage;
|
||||
import com.red.circle.mq.rocket.business.streams.GiveGiftSink;
|
||||
import com.red.circle.mq.rocket.business.streams.OfflineProcessGiftSink;
|
||||
import com.red.circle.tool.core.collection.CollectionUtils;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* MQ 礼物消息重试任务
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
@ConditionalOnProperty(name = "scheduler.gift-mq-retry", havingValue = "true", matchIfMissing = true)
|
||||
public class GiftMqRetryTask {
|
||||
|
||||
private final RocketMqFailMessageQueryService queryService;
|
||||
private final RocketMqFailMessageUpdateService updateService;
|
||||
private final GiftMqMessage giftMqMessage;
|
||||
|
||||
/**
|
||||
* 每分钟重试赠送礼物消息
|
||||
*/
|
||||
@Scheduled(cron = "0 */1 * * * ?")
|
||||
@TaskCacheLock(key = "GIFT_MQ_RETRY_TASK_GIVE_GIFT", expireSecond = 60)
|
||||
public void retryGiveGiftMessages() {
|
||||
try {
|
||||
List<RocketMqFailMessageDetail> messages = queryService.getPendingMessagesByTag(
|
||||
GiveGiftSink.TAG, 100);
|
||||
|
||||
if (CollectionUtils.isEmpty(messages)) {
|
||||
return;
|
||||
}
|
||||
|
||||
log.info("开始重试赠送礼物消息，count={}", messages.size());
|
||||
|
||||
for (RocketMqFailMessageDetail message : messages) {
|
||||
retryGiveGiftMessage(message);
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("重试赠送礼物消息异常", e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 每分钟重试离线统计礼物消息
|
||||
*/
|
||||
@Scheduled(cron = "0 */1 * * * ?")
|
||||
@TaskCacheLock(key = "GIFT_MQ_RETRY_TASK_OFFLINE_PROCESS", expireSecond = 60)
|
||||
public void retryOfflineProcessGiftMessages() {
|
||||
try {
|
||||
List<RocketMqFailMessageDetail> messages = queryService.getPendingMessagesByTag(
|
||||
OfflineProcessGiftSink.TAG, 100);
|
||||
|
||||
if (CollectionUtils.isEmpty(messages)) {
|
||||
return;
|
||||
}
|
||||
|
||||
log.info("开始重试离线统计礼物消息，count={}", messages.size());
|
||||
|
||||
for (RocketMqFailMessageDetail message : messages) {
|
||||
retryOfflineProcessMessage(message);
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("重试离线统计礼物消息异常", e);
|
||||
}
|
||||
}
|
||||
|
||||
private void retryGiveGiftMessage(RocketMqFailMessageDetail message) {
|
||||
String bizNo = message.getBizNo();
|
||||
|
||||
try {
|
||||
// 1. 加锁（防止并发重试）
|
||||
if (!updateService.lockForRetry(bizNo)) {
|
||||
return;
|
||||
}
|
||||
|
||||
// 2. 重新发送消息
|
||||
MessageEvent event = MessageEvent.builder()
|
||||
.consumeId(bizNo)
|
||||
.topic(FixedTopic.valueOf(message.getTopic()))
|
||||
.tag(message.getTag())
|
||||
.body(message.getMessageBody())
|
||||
.build();
|
||||
|
||||
if (message.getStartDeliverTime() != null) {
|
||||
event.setStartDeliverTime(message.getStartDeliverTime());
|
||||
}
|
||||
|
||||
giftMqMessage.sendGift(bizNo, parseGiveGiftEvent(message.getMessageBody()));
|
||||
|
||||
// 3. 标记成功
|
||||
updateService.markSuccess(bizNo);
|
||||
|
||||
log.info("赠送礼物消息重试成功，bizNo={}", bizNo);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("赠送礼物消息重试失败，bizNo={}, error={}", bizNo, e.getMessage(), e);
|
||||
updateService.markFailed(bizNo, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private void retryOfflineProcessMessage(RocketMqFailMessageDetail message) {
|
||||
String bizNo = message.getBizNo();
|
||||
|
||||
try {
|
||||
// 1. 加锁
|
||||
if (!updateService.lockForRetry(bizNo)) {
|
||||
return;
|
||||
}
|
||||
|
||||
// 2. 重新发送消息
|
||||
giftMqMessage.sendOfflineProcessGiftEvent(
|
||||
parseOfflineProcessEvent(message.getMessageBody()));
|
||||
|
||||
// 3. 标记成功
|
||||
updateService.markSuccess(bizNo);
|
||||
|
||||
log.info("离线统计礼物消息重试成功，bizNo={}", bizNo);
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("离线统计礼物消息重试失败，bizNo={}, error={}", bizNo, e.getMessage(), e);
|
||||
updateService.markFailed(bizNo, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private GiveAwayGiftBatchEvent parseGiveGiftEvent(
|
||||
String body) {
|
||||
return JSON.parseObject(body, GiveAwayGiftBatchEvent.class);
|
||||
}
|
||||
|
||||
private OfflineProcessGiftEvent parseOfflineProcessEvent(
|
||||
String body) {
|
||||
return JSON.parseObject(body, OfflineProcessGiftEvent.class);
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user