hy/aslan-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mq repeatConsumeTag 问题处理

Browse Source
This commit is contained in:
tianfeng 2025-12-24 10:42:15 +08:00
parent d1a4522d18
commit 37cd617687
2 changed files with 23 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
rc-service/rc-service-order/order-application/src/main/java/com/red/circle/order/app/listener/BuySuccessListener.java
View File

@ -1,5 +1,6 @@
package com.red.circle.order.app.listener; package com.red.circle.order.app.listener;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils; import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.red.circle.common.business.core.enums.ReceiptType; import com.red.circle.common.business.core.enums.ReceiptType;
import com.red.circle.common.business.core.enums.SysOriginPlatformEnum; import com.red.circle.common.business.core.enums.SysOriginPlatformEnum;
@ -71,7 +72,7 @@ public class BuySuccessListener implements MessageListener {
.logTag("语音房会话") .logTag("语音房会话")
.consumeMsgTimeoutMinute(15) .consumeMsgTimeoutMinute(15)
.repeatConsumeMinute(15) .repeatConsumeMinute(15)
.repeatConsumeTag("BuySuccess") .repeatConsumeTag(extractOrderNo(message))
.checkConsumeTag(BuySuccessSink.TAG) .checkConsumeTag(BuySuccessSink.TAG)
.message(message) .message(message)
.build(), .build(),
@ -79,6 +80,22 @@ public class BuySuccessListener implements MessageListener {
this::execute); this::execute);
} }
/**
* 提取订单号作为幂等键
*/
private String extractOrderNo(ConsumerMessage message) {
try {
BuySuccessEvent event = JSON.parseObject(
message.getBody(),
BuySuccessEvent.class
);
return "BUY_SUCCESS_" + event.getPurchaseHistoryId();
} catch (Exception e) {
log.warn("解析订单号失败，使用消息ID作为幂等键", e);
return "BUY_SUCCESS_" + message.getMessageId();
}
}
private void execute(BuySuccessEvent eventBody) { private void execute(BuySuccessEvent eventBody) {
if (!ArithmeticUtils.gt(eventBody.getCandyQuantity(), BigDecimal.ZERO) if (!ArithmeticUtils.gt(eventBody.getCandyQuantity(), BigDecimal.ZERO)
|| Objects.equals(eventBody.getTrialPeriod(), Boolean.TRUE)) { || Objects.equals(eventBody.getTrialPeriod(), Boolean.TRUE)) {

9
rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/listener/BadgeCheckListener.java
View File

@ -31,15 +31,16 @@ public class BadgeCheckListener implements MessageListener {
return messageEventProcess.consume( return messageEventProcess.consume(
MessageEventProcessDescribe.builder() MessageEventProcessDescribe.builder()
.logTag("徽章检测处理") .logTag("徽章检测处理")
.consumeMsgTimeoutMinute(60) .consumeMsgTimeoutMinute(5)
.repeatConsumeMinute(1) .repeatConsumeMinute(1)
.repeatConsumeTag("LiveVoiceSession") .repeatConsumeTag("BADGE_CHECK_" + message.getMessageId())
.checkConsumeTag(BadgeCheckSink.TAG) .checkConsumeTag(BadgeCheckSink.TAG)
.message(message) .message(message)
.build(), .build(),
BadgeEvent.class, BadgeEvent.class,
eventBody -> badgeCommon.checkActivateBadgePermanent(eventBody.getUserId(), eventBody -> {
BadgeKeyEnum.valueOf(eventBody.getBadgeKey()))); // badgeCommon.checkActivateBadgePermanent(eventBody.getUserId(),BadgeKeyEnum.valueOf(eventBody.getBadgeKey()));
});
} }