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

账单-雪花主键-mq-合并topic

Browse Source 
(cherry picked from commit e520926097)
This commit is contained in:
khalil
2025-09-11 14:01:41 +08:00
parent e92337882e
commit f09ba29175
6 changed files with 3 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
accompany-base/accompany-core/src/main/java/com/accompany/common/redis/RedisKey.java
View File

@@ -1394,7 +1394,6 @@ public enum RedisKey {
charge_floating_count, //充值飘屏计数器
bill_record_message,
lock_bill_record_message,
//小游戏重复订单

66
accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/gift/BillMessageService.java
View File

@@ -4,8 +4,6 @@ import com.accompany.business.message.BillMessage;
import com.accompany.business.service.mq.RocketMQService;
import com.accompany.business.service.user.UsersService;
import com.accompany.common.redis.RedisKey;
import com.accompany.common.status.BusiStatus;
import com.accompany.core.exception.ServiceException;
import com.accompany.core.model.Users;
import com.accompany.sharding.mapper.BillRecordMapper;
import com.accompany.sharding.model.BillRecord;
@@ -13,21 +11,13 @@ import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.incrementer.DefaultIdentifierGenerator;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@Slf4j
@Service
public class BillMessageService implements InitializingBean {
@@ -59,39 +49,6 @@ public class BillMessageService implements InitializingBean {
};
public void handleBillMessage(BillMessage billMessage) {
// 防止消息被重复消费
boolean locked = false;
RLock lock = recordMessMap.getLock(RedisKey.lock_bill_record_message.getKey(billMessage.getMessId()));
try {
locked = lock.tryLock(15, 5, TimeUnit.SECONDS);
if (!locked) {
log.warn("handleBillMessage billMessage had handle, mess: " + billMessage);
return;
}
if (!recordMessMap.containsKey(billMessage.getMessId())){
log.warn("handleBillMessage billMessage had handle, mess: " + billMessage);
return;
}
BillRecord billRecord = insertBillRecord(billMessage);
log.info("【处理账单mq】 billRecord 插入成功 id:{} messId: {} mess:{}",
billRecord.getBillId(), billMessage.getMessId(), JSON.toJSONString(billMessage));
recordMessMap.fastRemove(billMessage.getMessId());
} catch (InterruptedException e) {
log.error("handleBillMessage mess {} 持有所异常 ", JSON.toJSONString(billMessage), e);
throw new RuntimeException(e);
} finally {
if (locked) {
lock.unlock();
}
}
}
public void handleBillMessageV2(BillMessage billMessage) {
BillRecord billRecord = insertBillRecordIgnore(billMessage);
log.info("【处理账单mq】 billRecord 插入成功 id:{} messId: {} mess:{}",
@@ -100,29 +57,6 @@ public class BillMessageService implements InitializingBean {
recordMessMap.fastRemove(billMessage.getMessId());
}
private BillRecord insertBillRecord(BillMessage billMessage) {
BillRecord billRecord = new BillRecord();
BeanUtils.copyProperties(billMessage, billRecord);
Users u = usersService.getUsersByUid(billMessage.getUid());
if (null != u){
billRecord.setPartitionId(u.getPartitionId());
}
for (int i = 0; i < 3; i++) {
try {
billRecord.setBillId(null);
billRecordMapper.insert(billRecord);
return billRecord;
} catch (DuplicateKeyException ignore){
log.error("[insertBillRecord] 插入账单失败", ignore);
}
}
log.error(String.format("[insertBillRecord] 插入账单3次都失败 %s", JSON.toJSONString(billRecord)));
throw new ServiceException(BusiStatus.SERVERBUSY);
}
private BillRecord insertBillRecordIgnore(BillMessage billMessage) {
long startTime = System.currentTimeMillis();
long copyPropertiesTime = 0;

2
accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/mq/RocketMQService.java
View File

@@ -29,7 +29,7 @@ public class RocketMQService {
* 送消息，发送到MQ
*/
public void sendBillRecordMessage(BillMessage billMessage) {
mqMessageProducer.send(MqConstant.BILL_RECORD_V2_TOPIC, billMessage,
mqMessageProducer.send(MqConstant.BILL_RECORD_TOPIC, billMessage,
sendResult -> log.info("sendBillRecordMessage success message: {} queue {}", JSON.toJSONString(billMessage), sendResult.getMessageQueue().getQueueId()),
throwable -> log.error("sendBillRecordMessage fail message: {}", JSON.toJSONString(billMessage), throwable));
}

3
accompany-mq/accompany-mq-sdk/src/main/java/com/accompany/mq/constant/MqConstant.java
View File

@@ -35,9 +35,6 @@ public interface MqConstant {
String BILL_RECORD_TOPIC = "bill_record_topic";
String BILL_RECORD_CONSUME_GROUP = "bill_record_consume_group";
String BILL_RECORD_V2_TOPIC = "bill_record_v2_topic";
String BILL_RECORD_V2_CONSUME_GROUP = "bill_record_v2_consume_group";
String BRAVO_TOPIC = "bravo_topic";
String BRAVO_CONSUME_GROUP = "bravo_consume_group";

28
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/BillMessageV2Consumer.java
View File

@@ -1,28 +0,0 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.BillMessage;
import com.accompany.business.service.gift.BillMessageService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.BILL_RECORD_V2_TOPIC, consumerGroup = MqConstant.BILL_RECORD_V2_CONSUME_GROUP)
public class BillMessageV2Consumer extends AbstractMessageListener<BillMessage> {
@Autowired
private BillMessageService billMessageService;
@Override
public void onMessage(BillMessage billMessage) {
log.info("onMessage billMessage v2: {}", billMessage.toString());
billMessageService.handleBillMessageV2(billMessage);
}
}

4
accompany-scheduler/accompany-scheduler-service/src/main/java/com/accompany/scheduler/task/BillTask.java
View File

@@ -24,7 +24,7 @@ public class BillTask extends BaseTask {
@Scheduled(cron = "0 */5 * * * ?")
public void retryBillQueue() {
log.info("retryBillQueue start ...");
Map<String, BillMessage> map = billMessageService.getRecordMessMap().randomEntries(20000);
Map<String, BillMessage> map = billMessageService.getRecordMessMap().randomEntries(10000);
if (CollectionUtils.isEmpty(map)) {
return;
}
@@ -36,7 +36,7 @@ public class BillTask extends BaseTask {
String messId = entry.getKey();
BillMessage val = entry.getValue();
if (curTime - val.getCreateTime().getTime() > gapTime) {
billMessageService.handleBillMessageV2(val);
billMessageService.handleBillMessage(val);
}
} catch (Exception e) {
log.error("retryBillQueue error", e);