From d2f00bb7f84c6446c3cc214d2c13b673f9392370 Mon Sep 17 00:00:00 2001
From: khalil <842328916@qq.com>
Date: Tue, 9 Sep 2025 15:54:05 +0800
Subject: [PATCH] =?UTF-8?q?=E8=B4=A6=E5=8D=95-=E9=9B=AA=E8=8A=B1=E4=B8=BB?=
 =?UTF-8?q?=E9=94=AE-=E6=97=A0=E9=94=81=E5=8C=96=E5=94=AF=E4=B8=80?=
 =?UTF-8?q?=E4=B8=BB=E9=94=AE=E6=8F=92=E5=85=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../sharding/mapper/BillRecordMapper.java     |  4 +-
 .../sharding/sqlmappers/BillRecordMapper.xml  | 82 +++++++++----------
 .../service/gift/BillMessageService.java      | 57 ++++++++-----
 .../business/service/mq/RocketMQService.java  |  2 +-
 .../com/accompany/mq/constant/MqConstant.java |  3 +
 .../mq/consumer/BillMessageConsumer.java      |  1 -
 .../mq/consumer/BillMessageV2Consumer.java    | 28 +++++++
 7 files changed, 109 insertions(+), 68 deletions(-)
 create mode 100644 accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/BillMessageV2Consumer.java

diff --git a/accompany-base/accompany-sharding/accompany-sharding-service/src/main/java/com/accompany/sharding/mapper/BillRecordMapper.java b/accompany-base/accompany-sharding/accompany-sharding-service/src/main/java/com/accompany/sharding/mapper/BillRecordMapper.java
index 1cff1c731..8ab4cbf8e 100644
--- a/accompany-base/accompany-sharding/accompany-sharding-service/src/main/java/com/accompany/sharding/mapper/BillRecordMapper.java
+++ b/accompany-base/accompany-sharding/accompany-sharding-service/src/main/java/com/accompany/sharding/mapper/BillRecordMapper.java
@@ -21,10 +21,8 @@ public interface BillRecordMapper extends BaseMapper<BillRecord> {
 
     /**
      * 批量插入账单记录，使用 INSERT IGNORE 忽略重复记录
-     * @param billRecords 账单记录列表
-     * @return 插入的记录数
      */
-    int insertIgnore(@Param("billRecords") List<BillRecord> billRecords);
+    int insertIgnore(@Param("record") BillRecord billRecord);
 
     Long getGiftSumByGift();
 
diff --git a/accompany-base/accompany-sharding/accompany-sharding-service/src/main/resources/sharding/sqlmappers/BillRecordMapper.xml b/accompany-base/accompany-sharding/accompany-sharding-service/src/main/resources/sharding/sqlmappers/BillRecordMapper.xml
index 66f2f4197..dcacda4ef 100644
--- a/accompany-base/accompany-sharding/accompany-sharding-service/src/main/resources/sharding/sqlmappers/BillRecordMapper.xml
+++ b/accompany-base/accompany-sharding/accompany-sharding-service/src/main/resources/sharding/sqlmappers/BillRecordMapper.xml
@@ -15,49 +15,47 @@
 
     <insert id="insertIgnore">
         INSERT IGNORE INTO bill_record (
-            bill_id,
-            uid,
-            partition_id,
-            target_uid,
-            room_uid,
-            bill_type,
-            obj_id,
-            obj_type,
-            gift_id,
-            gift_num,
-            gift_total_gold_num,
-            currency,
-            before_amount,
-            amount,
-            actual_amount,
-            after_amount,
-            create_time,
-            remark,
-            mess_id
+        bill_id,
+        uid,
+        partition_id,
+        target_uid,
+        room_uid,
+        bill_type,
+        obj_id,
+        obj_type,
+        gift_id,
+        gift_num,
+        gift_total_gold_num,
+        currency,
+        before_amount,
+        amount,
+        actual_amount,
+        after_amount,
+        create_time,
+        remark,
+        mess_id
         ) VALUES
-        <foreach collection="billRecords" item="record" separator=",">
-            (
-                #{record.billId},
-                #{record.uid},
-                #{record.partitionId},
-                #{record.targetUid},
-                #{record.roomUid},
-                #{record.billType},
-                #{record.objId},
-                #{record.objType},
-                #{record.giftId},
-                #{record.giftNum},
-                #{record.giftTotalGoldNum},
-                #{record.currency},
-                #{record.beforeAmount},
-                #{record.amount},
-                #{record.actualAmount},
-                #{record.afterAmount},
-                #{record.createTime},
-                #{record.remark},
-                #{record.messId}
-            )
-        </foreach>
+        (
+        #{record.billId},
+        #{record.uid},
+        #{record.partitionId},
+        #{record.targetUid},
+        #{record.roomUid},
+        #{record.billType},
+        #{record.objId},
+        #{record.objType},
+        #{record.giftId},
+        #{record.giftNum},
+        #{record.giftTotalGoldNum},
+        #{record.currency},
+        #{record.beforeAmount},
+        #{record.amount},
+        #{record.actualAmount},
+        #{record.afterAmount},
+        #{record.createTime},
+        #{record.remark},
+        #{record.messId}
+        )
     </insert>
 
     <select id="getGiftSumByGift" resultType="long">
diff --git a/accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/gift/BillMessageService.java b/accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/gift/BillMessageService.java
index 3c2a4f4b3..bbcd9011c 100644
--- a/accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/gift/BillMessageService.java
+++ b/accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/gift/BillMessageService.java
@@ -91,29 +91,13 @@ public class BillMessageService implements InitializingBean {
         }
     }
 
-    public void handleBillMessage(List<BillMessage> billMessageList) {
-        List<Long> uidList = billMessageList.stream().map(BillMessage::getUid).distinct().toList();
-        Map<Long, Users> usersMap = usersService.getUsersMapByUids(uidList);
+    public void handleBillMessageV2(BillMessage billMessage) {
 
-        List<BillRecord> poList = billMessageList.stream().map(billMessage -> {
-            BillRecord billRecord = new BillRecord();
-            BeanUtils.copyProperties(billMessage, billRecord);
-            Users u = usersMap.get(billMessage.getUid());
-            if (null != u){
-                billRecord.setPartitionId(u.getPartitionId());
-            }
+        BillRecord billRecord = insertBillRecordIgnore(billMessage);
+        log.info("【处理账单mq】 billRecord 插入成功 id:{} messId: {} mess:{}",
+                billRecord.getBillId(), billMessage.getMessId(), JSON.toJSONString(billMessage));
 
-            if (null == billRecord.getBillId()){
-                billRecord.setBillId(DefaultIdentifierGenerator.getInstance().nextId(null));
-            }
-
-            return billRecord;
-        }).toList();
-
-        billRecordMapper.insertIgnore(poList);
-
-        String[] messIdList = billMessageList.stream().map(BillMessage::getMessId).toArray(String[]::new);
-        recordMessMap.fastRemove(messIdList);
+        recordMessMap.fastRemove(billMessage.getMessId());
     }
 
     private BillRecord insertBillRecord(BillMessage billMessage) {
@@ -139,6 +123,37 @@ public class BillMessageService implements InitializingBean {
         throw new ServiceException(BusiStatus.SERVERBUSY);
     }
 
+    private BillRecord insertBillRecordIgnore(BillMessage billMessage) {
+        long startTime = System.currentTimeMillis();
+        long copyPropertiesTime = 0;
+        long getUserTime = 0;
+        long insertTime = 0;
+
+        BillRecord billRecord = new BillRecord();
+        BeanUtils.copyProperties(billMessage, billRecord);
+        copyPropertiesTime = System.currentTimeMillis();
+
+        Users u = usersService.getUsersByUid(billMessage.getUid());
+        getUserTime = System.currentTimeMillis();
+        if (null != u){
+            billRecord.setPartitionId(u.getPartitionId());
+        }
+
+        int insertRow = billRecordMapper.insertIgnore(billRecord);
+        insertTime = System.currentTimeMillis();
+
+        long endTime = System.currentTimeMillis();
+        
+        log.info("insertBillRecordIgnore row {} performance - copy: {}ms, getUser: {}ms, insert: {}ms, total: {}ms",
+                insertRow,
+                copyPropertiesTime - startTime,
+                getUserTime - copyPropertiesTime,
+                insertTime - getUserTime,
+                endTime - startTime);
+
+        return billRecord;
+    }
+
     @Override
     public void afterPropertiesSet() throws Exception {
         recordMessMap = redissonClient.getMap(RedisKey.bill_record_message.getKey());
diff --git a/accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/mq/RocketMQService.java b/accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/mq/RocketMQService.java
index 0a91641bf..11bb323ec 100644
--- a/accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/mq/RocketMQService.java
+++ b/accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/mq/RocketMQService.java
@@ -29,7 +29,7 @@ public class RocketMQService {
      * 送消息，发送到MQ
      */
     public void sendBillRecordMessage(BillMessage billMessage) {
-        mqMessageProducer.send(MqConstant.BILL_RECORD_TOPIC, billMessage,
+        mqMessageProducer.send(MqConstant.BILL_RECORD_V2_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));
     }
diff --git a/accompany-mq/accompany-mq-sdk/src/main/java/com/accompany/mq/constant/MqConstant.java b/accompany-mq/accompany-mq-sdk/src/main/java/com/accompany/mq/constant/MqConstant.java
index 0ffc9f161..f66014730 100644
--- a/accompany-mq/accompany-mq-sdk/src/main/java/com/accompany/mq/constant/MqConstant.java
+++ b/accompany-mq/accompany-mq-sdk/src/main/java/com/accompany/mq/constant/MqConstant.java
@@ -35,6 +35,9 @@ 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";
 
diff --git a/accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/BillMessageConsumer.java b/accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/BillMessageConsumer.java
index decde4f3b..998e9a01b 100644
--- a/accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/BillMessageConsumer.java
+++ b/accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/BillMessageConsumer.java
@@ -5,7 +5,6 @@ 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.ConsumeMode;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
diff --git a/accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/BillMessageV2Consumer.java b/accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/BillMessageV2Consumer.java
new file mode 100644
index 000000000..65f2a6386
--- /dev/null
+++ b/accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/BillMessageV2Consumer.java
@@ -0,0 +1,28 @@
+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);
+    }
+
+}