账单-雪花主键-mq批量消费-sql

accompany-base/accompany-sharding/accompany-sharding-service/src/main/java/com/accompany/sharding/mapper/BillRecordMapper.java

@@ -19,6 +19,13 @@ import java.util.Map;
  */
 public interface BillRecordMapper extends BaseMapper<BillRecord> {
 
+    /**
+     * 批量插入账单记录，使用 INSERT IGNORE 忽略重复记录
+     * @param billRecords 账单记录列表
+     * @return 插入的记录数
+     */
+    int insertIgnore(@Param("billRecords") List<BillRecord> billRecords);
+
     Long getGiftSumByGift();
 
     Long getGiftTypeSumByRoom();

accompany-base/accompany-sharding/accompany-sharding-service/src/main/resources/sharding/sqlmappers/BillRecordMapper.xml

@@ -13,6 +13,53 @@
         <result column="create_time" property="createTime" jdbcType="TIMESTAMP"/>
     </resultMap>
 
+    <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
+        ) 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>
+    </insert>
+
     <select id="getGiftSumByGift" resultType="long">
     select COUNT(bill_id) FROM bill_record
   </select>

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

@@ -22,7 +22,11 @@ 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
@@ -87,6 +91,31 @@ 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);
+
+        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());
+            }
+
+            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);
+    }
+
     private BillRecord insertBillRecord(BillMessage billMessage) {
         BillRecord billRecord = new BillRecord();
         BeanUtils.copyProperties(billMessage, billRecord);