From 2a05a4ad0089e6561483accbe8bf8c3bd6630dda Mon Sep 17 00:00:00 2001 From: tianfeng <769204422@qq.com> Date: Mon, 5 Jan 2026 17:24:21 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9Emq=E6=B6=88=E6=81=AF=E5=8F=91?= =?UTF-8?q?=E9=80=81=E5=A4=B1=E8=B4=A5=E9=87=8D=E8=AF=95=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...mmonBusinessRocketMqAutoConfiguration.java | 1 + .../RocketMqFailMessageDetail.java | 60 +++++++ .../failmessage/RocketMqFailMessageKeys.java | 33 ++++ .../RocketMqFailMessageQueryService.java | 26 +++ .../RocketMqFailMessageRetryStrategy.java | 44 +++++ .../RocketMqFailMessageUpdateService.java | 38 +++++ .../RocketMqFailMessageQueryServiceImpl.java | 65 ++++++++ .../RocketMqFailMessageUpdateServiceImpl.java | 155 ++++++++++++++++++ .../RocketMqSenderFailServiceRedisImpl.java | 74 +++++++++ .../other/app/scheduler/GiftMqRetryTask.java | 154 +++++++++++++++++ 10 files changed, 650 insertions(+) create mode 100644 rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageDetail.java create mode 100644 rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageKeys.java create mode 100644 rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageQueryService.java create mode 100644 rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageRetryStrategy.java create mode 100644 rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageUpdateService.java create mode 100644 rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqFailMessageQueryServiceImpl.java create mode 100644 rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqFailMessageUpdateServiceImpl.java create mode 100644 rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqSenderFailServiceRedisImpl.java create mode 100644 rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/GiftMqRetryTask.java diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/CommonBusinessRocketMqAutoConfiguration.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/CommonBusinessRocketMqAutoConfiguration.java index f80bd459..d41be370 100644 --- a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/CommonBusinessRocketMqAutoConfiguration.java +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/CommonBusinessRocketMqAutoConfiguration.java @@ -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" }) diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageDetail.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageDetail.java new file mode 100644 index 00000000..76261f71 --- /dev/null +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageDetail.java @@ -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; +} diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageKeys.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageKeys.java new file mode 100644 index 00000000..564a8d35 --- /dev/null +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageKeys.java @@ -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; + } +} diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageQueryService.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageQueryService.java new file mode 100644 index 00000000..61eff4b7 --- /dev/null +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageQueryService.java @@ -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 getPendingMessagesByTag(String tag, int limit); + + /** + * 根据业务号获取消息详情 + * + * @param bizNo 业务唯一标识 + * @return 消息详情 + */ + RocketMqFailMessageDetail getMessageByBizNo(String bizNo); +} diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageRetryStrategy.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageRetryStrategy.java new file mode 100644 index 00000000..fdaa3f95 --- /dev/null +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageRetryStrategy.java @@ -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); + } +} diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageUpdateService.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageUpdateService.java new file mode 100644 index 00000000..cc86b9a9 --- /dev/null +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/RocketMqFailMessageUpdateService.java @@ -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); +} diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqFailMessageQueryServiceImpl.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqFailMessageQueryServiceImpl.java new file mode 100644 index 00000000..06e0fe00 --- /dev/null +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqFailMessageQueryServiceImpl.java @@ -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 getPendingMessagesByTag(String tag, int limit) { + try { + // 1. 从 Set 中获取该 tag 的所有 bizNo + String tagGroupKey = RocketMqFailMessageKeys.getTagGroupKey(tag); + Set 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; + } + } +} diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqFailMessageUpdateServiceImpl.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqFailMessageUpdateServiceImpl.java new file mode 100644 index 00000000..5a9bd248 --- /dev/null +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqFailMessageUpdateServiceImpl.java @@ -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); + } + } +} diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqSenderFailServiceRedisImpl.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqSenderFailServiceRedisImpl.java new file mode 100644 index 00000000..71c13188 --- /dev/null +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/failmessage/impl/RocketMqSenderFailServiceRedisImpl.java @@ -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); + } + } +} diff --git a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/GiftMqRetryTask.java b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/GiftMqRetryTask.java new file mode 100644 index 00000000..7873107b --- /dev/null +++ b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/scheduler/GiftMqRetryTask.java @@ -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 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 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); + } +}