使用先查后改解决批量插入在并发时的死锁问题；

设备95值保存失败的原因是数据库mac_address字段长度不足，改为500后正常保存。

tongran-modules/tongran-system/src/main/java/com/tongran/system/mapper/AllInterfaceNameMapper.java

@@ -99,4 +99,19 @@ public interface AllInterfaceNameMapper
      * @return
      */
     Integer countNetChild(AllInterfaceName allInterfaceName);
+
+    /**
+     * 按唯一键查询已存在的记录
+     */
+    List<String> selectExisting(@Param("list") List<AllInterfaceName> list);
+
+    /**
+     * 纯批量插入（无 ON DUPLICATE KEY UPDATE）
+     */
+    int batchInsertPure(@Param("list") List<AllInterfaceName> list);
+
+    /**
+     * 按唯一键批量更新
+     */
+    int batchUpdateByUniqueKey(@Param("list") List<AllInterfaceName> list);
 }

tongran-modules/tongran-system/src/main/java/com/tongran/system/service/impl/EpsServerRevenueConfigServiceImpl.java

@@ -15,10 +15,7 @@ import org.springframework.transaction.annotation.Isolation;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.CollectionUtils;
 
-import java.util.ArrayList;
-import java.util.Comparator;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 /**
  * 服务器收益方式配置Service业务层处理
@@ -379,8 +376,38 @@ public class EpsServerRevenueConfigServiceImpl implements IEpsServerRevenueConfi
     private String getCompositeKey(String clientId, String interfaceName) {
         return clientId + "|" + interfaceName;
     }
+//    /**
+//     * 批量处理接口名称
+//     */
+//    @Transactional(rollbackFor = Exception.class, isolation = Isolation.READ_COMMITTED)
+//    public void processInterfaceNames(List<EpsInitialTrafficData> trafficDataList) {
+//        if (CollectionUtils.isEmpty(trafficDataList)) {
+//            return;
+//        }
+//
+//        // 直接转换并过滤
+//        List<AllInterfaceName> records = new ArrayList<>();
+//        for (EpsInitialTrafficData data : trafficDataList) {
+//            if (isValidData(data)) {
+//                records.add(convertToInterfaceName(data));
+//            }
+//        }
+//
+//        if (!records.isEmpty()) {
+//            // 排序防止死锁
+//            records.sort(Comparator
+//                    .comparing(AllInterfaceName::getInterfaceName)
+//                    .thenComparing(AllInterfaceName::getClientId)
+//                    .thenComparing(AllInterfaceName::getServerIp)
+//                    .thenComparing(AllInterfaceName::getResourceType));
+//
+//            allInterfaceNameMapper.batchInsert(records);
+//            log.info("批量处理完成，记录数: {}", records.size());
+//        }
+//    }
+
     /**
-     * 批量处理接口名称
+     * 批量处理接口名称（不排序，先查后插，彻底避免死锁）
      */
     @Transactional(rollbackFor = Exception.class, isolation = Isolation.READ_COMMITTED)
     public void processInterfaceNames(List<EpsInitialTrafficData> trafficDataList) {
@@ -388,7 +415,7 @@ public class EpsServerRevenueConfigServiceImpl implements IEpsServerRevenueConfi
             return;
         }
 
-        // 直接转换并过滤
+        // 1. 转换并过滤
         List<AllInterfaceName> records = new ArrayList<>();
         for (EpsInitialTrafficData data : trafficDataList) {
             if (isValidData(data)) {
@@ -396,18 +423,46 @@ public class EpsServerRevenueConfigServiceImpl implements IEpsServerRevenueConfi
             }
         }
 
-        if (!records.isEmpty()) {
-            // 排序防止死锁
-            records.sort(Comparator
-                    .comparing(AllInterfaceName::getInterfaceName)
-                    .thenComparing(AllInterfaceName::getClientId)
-                    .thenComparing(AllInterfaceName::getServerIp)
-                    .thenComparing(AllInterfaceName::getResourceType));
-
-            allInterfaceNameMapper.batchInsert(records);
-            log.info("批量处理完成，记录数: {}", records.size());
+        if (records.isEmpty()) {
+            return;
         }
+
+        // 2. 先查出数据库里已存在的记录（按唯一键 uk_interface_unified_ip）
+        List<String> existingKeys = allInterfaceNameMapper.selectExisting(records);
+        Set<String> existingSet = new HashSet<>(existingKeys);
+
+        // 3. 分离 insert 和 update
+        List<AllInterfaceName> toInsert = new ArrayList<>();
+        List<AllInterfaceName> toUpdate = new ArrayList<>();
+
+        for (AllInterfaceName record : records) {
+            // 统一按生成列逻辑计算 unified_ip
+            String unifiedIp = StringUtils.isNotBlank(record.getServerIp())
+                    ? record.getServerIp()
+                    : record.getSwitchIp();
+
+            String key = record.getInterfaceName() + "#" + record.getClientId()
+                    + "#" + unifiedIp + "#" + record.getResourceType();
+
+            if (existingSet.contains(key)) {
+                toUpdate.add(record);
+            } else {
+                toInsert.add(record);
+            }
+        }
+
+        // 4. 分别执行
+        if (!toInsert.isEmpty()) {
+            allInterfaceNameMapper.batchInsertPure(toInsert);
+        }
+        if (!toUpdate.isEmpty()) {
+            allInterfaceNameMapper.batchUpdateByUniqueKey(toUpdate);
+        }
+
+        log.info("批量处理完成，insert={}, update={}, total={}",
+                toInsert.size(), toUpdate.size(), records.size());
     }
+
     private boolean isValidData(EpsInitialTrafficData data) {
         return StringUtils.isNotBlank(data.getName())
                 && StringUtils.isNotBlank(data.getMac())

tongran-modules/tongran-system/src/main/java/com/tongran/system/service/impl/RmResourceRegistrationServiceImpl.java

@@ -1121,7 +1121,7 @@ public class RmResourceRegistrationServiceImpl implements IRmResourceRegistratio
         Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒
         String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss", createTime);
         List<NetworkInfo> networkInfoList = registerMsg.getNetworkInfo();
-        if(!networkInfoList.isEmpty()){
+        if(networkInfoList != null && !networkInfoList.isEmpty()){
             // 查询此数据是否存在
             RmResourceRegistration queryParam = new RmResourceRegistration();
             queryParam.setClientId(registerMsg.getClientId());

tongran-modules/tongran-system/src/main/resources/mapper/system/AllInterfaceNameMapper.xml

@@ -320,4 +320,65 @@
         select count(1) from rm_network_interface_child
         where client_id = #{clientId} and parent_interface = SUBSTRING_INDEX(#{interfaceName}, '(', 1)
     </select>
+
+    <!-- 按唯一键查询已存在的记录，返回拼接好的 key -->
+    <select id="selectExisting" resultType="java.lang.String">
+        SELECT CONCAT(interface_name, '#', client_id, '#', unified_ip, '#', resource_type)
+        FROM all_interface_name
+        WHERE (interface_name, client_id, unified_ip, resource_type) IN
+        <foreach collection="list" item="item" open="(" separator="," close=")">
+            (
+            #{item.interfaceName},
+            #{item.clientId},
+            COALESCE(#{item.serverIp}, #{item.switchIp}),
+            #{item.resourceType}
+            )
+        </foreach>
+    </select>
+
+    <!-- 纯批量插入（无 ON DUPLICATE KEY UPDATE） -->
+    <insert id="batchInsertPure" parameterType="java.util.List">
+        INSERT INTO all_interface_name
+        (
+        interface_name, client_id, resource_type,
+        device_sn, node_name, business_code, business_name,
+        switch_name, interface_device_type, server_port,
+        switch_sn, switch_ip, server_ip, port_status,
+        create_time, update_time
+        )
+        VALUES
+        <foreach collection="list" item="item" separator=",">
+            (
+            #{item.interfaceName}, #{item.clientId}, #{item.resourceType},
+            #{item.deviceSn}, #{item.nodeName}, #{item.businessCode}, #{item.businessName},
+            #{item.switchName}, #{item.interfaceDeviceType}, #{item.serverPort},
+            #{item.switchSn}, #{item.switchIp}, #{item.serverIp}, #{item.portStatus},
+            NOW(), NOW()
+            )
+        </foreach>
+    </insert>
+
+    <!-- 按唯一键批量更新（需要 allowMultiQueries=true） -->
+    <update id="batchUpdateByUniqueKey" parameterType="java.util.List">
+        <foreach collection="list" item="item" separator=";">
+            UPDATE all_interface_name
+            SET
+            device_sn = #{item.deviceSn},
+            node_name = #{item.nodeName},
+            business_code = #{item.businessCode},
+            business_name = #{item.businessName},
+            switch_name = #{item.switchName},
+            interface_device_type = #{item.interfaceDeviceType},
+            server_port = #{item.serverPort},
+            switch_sn = #{item.switchSn},
+            switch_ip = #{item.switchIp},
+            server_ip = #{item.serverIp},
+            port_status = #{item.portStatus},
+            update_time = NOW()
+            WHERE interface_name = #{item.interfaceName}
+            AND client_id = #{item.clientId}
+            AND unified_ip = COALESCE(#{item.serverIp}, #{item.switchIp})
+            AND resource_type = #{item.resourceType}
+        </foreach>
+    </update>
 </mapper>

tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/RocketMsgListener.java

@@ -2,6 +2,7 @@ package com.tongran.rocketmq.consumer;
 
 import com.alibaba.fastjson.JSON;
 import com.tongran.common.core.enums.MsgEnum;
+import com.tongran.common.core.utils.ExceptionUtil;
 import com.tongran.rocketmq.domain.DeviceMessage;
 import com.tongran.rocketmq.enums.MessageCodeEnum;
 import com.tongran.rocketmq.handler.DeviceMessageHandler;
@@ -41,7 +42,8 @@ public class RocketMsgListener implements MessageListenerConcurrently {
      */
     @Override
     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
-       try{
+        String msgId = "";
+        try{
            //消息不等于空情况
            if (!CollectionUtils.isEmpty(list)) {
                //获取topic
@@ -49,11 +51,11 @@ public class RocketMsgListener implements MessageListenerConcurrently {
                    // 解析消息内容
                    // 明确指定UTF-8编码
                    String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
-                   log.info("接受到的消息为：{}", body);
                    String tags = messageExt.getTags();
                    String topic = messageExt.getTopic();
-                   String msgId = messageExt.getMsgId();
+                   msgId = messageExt.getMsgId();
                    String keys = messageExt.getKeys();
+                   log.info("接收到消息, msgId={}, body={}", msgId, body);
                    int reConsume = messageExt.getReconsumeTimes();
                    // 消息已经重试了3次，如果不需要再次消费，则返回成功
                    if (reConsume == 3) {
@@ -61,14 +63,14 @@ public class RocketMsgListener implements MessageListenerConcurrently {
                        DeviceMessage message = JSON.parseObject(body, DeviceMessage.class);
                        if(!message.getDataType().equals(MsgEnum.网络上报重试.getValue())){
                            // TODO 补偿信息
-                           log.error("消息消费三次失败，消息内容：{}", body);
+                           log.error("消息消费三次失败，msgId={}, body={}", body);
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//根据业务返回是否正常
                        }
                    }
                    // 流量数据重试
                    if (reConsume == 6) {
                        // 补偿信息
-                       log.error("流量数据重试消息消费6次失败，消息内容：{}", body);
+                       log.error("流量数据重试消息消费六次失败，msgId={}, body={}", body);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//根据业务返回是否正常
                    }
                    if(MessageCodeEnum.TONGRAN_AGENT_UP.getCode().equals(topic)){
@@ -93,7 +95,7 @@ public class RocketMsgListener implements MessageListenerConcurrently {
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        } catch (Exception e) {
            // 调用 handleException 方法处理异常并返回处理结果
-           return handleException(e);
+           return handleException(msgId, e);
        }
     }
 
@@ -103,14 +105,14 @@ public class RocketMsgListener implements MessageListenerConcurrently {
      * @param e 捕获的异常
      * @return 消息消费结果
      */
-    private static ConsumeConcurrentlyStatus handleException(final Exception e) {
+    private static ConsumeConcurrentlyStatus handleException(String msgId, final Exception e) {
         Class exceptionClass = e.getClass();
         if (exceptionClass.equals(UnsupportedEncodingException.class)) {
-            log.error(e.getMessage());
+            log.error("消息处理失败, msgId={}, error={}", msgId, e.getMessage());
         } else if (exceptionClass.equals(ConsumeException.class)) {
-            log.error(e.getMessage());
+            log.error("消息处理失败, msgId={}, error={}", msgId, e.getMessage());
         } else{
-            log.error(e.getMessage());
+            log.error("消息处理失败, msgId={}, error={} {}", msgId, e.getMessage(), ExceptionUtil.getExceptionMessage(e));
         }
         return ConsumeConcurrentlyStatus.RECONSUME_LATER;
     }

tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java

@@ -768,7 +768,7 @@ public class MessageHandler {
             // 自动注册服务器信息
             RmRegisterMsgRemote rmRegisterMsgRemote = new RmRegisterMsgRemote();
             BeanUtils.copyProperties(registerMsg, rmRegisterMsgRemote);
-            int rows = remoteRevenueConfigService.innerAddRegist(rmRegisterMsgRemote, SecurityConstants.INNER).getData();
+            int rows = Optional.ofNullable(remoteRevenueConfigService.innerAddRegist(rmRegisterMsgRemote, SecurityConstants.INNER).getData()).orElse(0);
             if(rows == 2){
                 // 注册成功，下发优先级为0的策略
                 rmMonitorPolicyService.issueDefaultPolicyByClientId(message.getClientId());
@@ -1963,7 +1963,7 @@ public class MessageHandler {
         Date createTime = new Date(millis / 1000 * 1000);
 
         List<NetworkInfo> networkInfoList = registerMsg.getNetworkInfo();
-        if (networkInfoList.isEmpty()) {
+        if (networkInfoList == null || networkInfoList.isEmpty()) {
             return;
         }