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

设备95值保存失败的原因是数据库mac_address字段长度不足，改为500后正常保存。
2026-07-27 10:53:21 +08:00
parent 368121b938
commit 92571fed91
4 changed files with 148 additions and 17 deletions
@@ -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);
 }
@@ -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())
@@ -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());
@@ -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>