moliparty/peko-java
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rocketmq-AbstractMessageListener去掉通用分布式锁

Browse Source
This commit is contained in:
khalil
2025-09-09 15:52:34 +08:00
parent faa924328b
commit 53f93c04a0
6 changed files with 25 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
accompany-mq/accompany-mq-service/src/main/java/com/accompany/mq/listener/AbstractMessageListener.java
View File

@@ -1,17 +1,14 @@
package com.accompany.mq.listener;
import cn.hutool.core.util.StrUtil;
import com.accompany.common.redis.RedisKey;
import com.accompany.core.service.common.JedisLockService;
import com.accompany.core.service.common.JedisService;
import com.accompany.common.status.BusiStatus;
import com.accompany.core.exception.ServiceException;
import com.accompany.mq.model.BaseMqMessage;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import java.lang.reflect.ParameterizedType;
import java.util.UUID;
/**
* @author: liaozetao
@@ -21,44 +18,21 @@ import java.util.UUID;
@Slf4j
public abstract class AbstractMessageListener<T extends BaseMqMessage> implements RocketMQListener<String> {
private static final int MQ_LOCK_SECONDS = 30 * 60;
@Autowired
protected JedisService jedisService;
@Autowired
protected JedisLockService jedisLockService;
@Override
public void onMessage(String message) {
try {
//log.info("====mq message start====");
//log.info("text message : {}", message);
if (!message.startsWith(StrUtil.DELIM_START) || !message.endsWith(StrUtil.DELIM_END)) {
return;
}
T mqMessage = JSONObject.parseObject(message)
.toJavaObject(((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0]);
if (mqMessage == null) {
return;
}
//String messId = mqMessage.getMessId();
String messId = UUID.randomUUID().toString();
//防止消息被重复消费
RedisKey mqLock = mqLock();
if (mqLock != null && !jedisService.setnx(mqLock.getKey(messId), Boolean.TRUE.toString(), MQ_LOCK_SECONDS)) {
log.error("mq lock : {}, message had handle, msg : {}", mqLock.getKey(messId), message);
return;
}
onMessage(mqMessage);
} catch (Exception e) {
log.error(e.getMessage(), e);
//log.info("====mq message start====");
//log.info("text message : {}", message);
if (!message.startsWith(StrUtil.DELIM_START) || !message.endsWith(StrUtil.DELIM_END)) {
return;
}
T mqMessage = JSONObject.parseObject(message)
.toJavaObject(((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0]);
if (mqMessage == null) {
throw new ServiceException(BusiStatus.PARAMERROR);
}
onMessage(mqMessage);
}
protected abstract void onMessage(T object) throws Exception;
protected abstract void onMessage(T object);
protected RedisKey mqLock() {
return null;
}
}

5
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/MyCardChargeMessageConsumer.java
View File

@@ -4,6 +4,7 @@ import cn.hutool.core.util.StrUtil;
import com.accompany.business.service.mycard.MyCardBizService;
import com.accompany.common.constant.Constant;
import com.accompany.common.redis.RedisKey;
import com.accompany.core.service.common.JedisService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import com.accompany.mq.model.ChargeMqMessage;
@@ -36,9 +37,11 @@ public class MyCardChargeMessageConsumer extends AbstractMessageListener<ChargeM
@Autowired
private MQMessageProducer mqMessageProducer;
@Autowired
private JedisService jedisService;
@Override
protected void onMessage(ChargeMqMessage object) throws Exception {
protected void onMessage(ChargeMqMessage object) {
String chargeRecordId = object.getChargeRecordId();
ChargeRecord chargeRecord = chargeRecordService.getChargeRecordById(chargeRecordId);
if (chargeRecord == null) {