refactor(utils): 优化设备离线检查任务的并发安全和状态管理

- 添加在线设备集合跟踪已成功同步在线的设备状态
- 引入设备级别锁机制防止并发访问冲突
- 实现设备锁的哈希分配算法确保线程安全
- 优化离线检查逻辑避免旧快照覆盖新上报数据
- 改进设备上线流程支持首次上报和服务重启场景
- 添加设备代码空值校验防止空指针异常
- 统一多个服务中的设备离线检查任务实现
- 在manhole服务中启用负载均衡RestTemplate配置

environment-service/src/main/java/com/zksy/environment/utils/DeviceOfflineCheckTask.java

@@ -12,6 +12,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestTemplate;
 
 import java.util.Date;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -29,13 +30,20 @@ public class DeviceOfflineCheckTask {
 
     /** 已知离线设备集合，用于避免重复API调用 */
     private final Set<String> offlineDeviceSet = ConcurrentHashMap.newKeySet();
+    /** 本实例已成功同步在线的设备；不能把尚未检测过离线当作已经在线。 */
+    private final Set<String> onlineDeviceSet = ConcurrentHashMap.newKeySet();
+    private final Object[] deviceLocks = new Object[128];
+
+    public DeviceOfflineCheckTask() {
+        Arrays.setAll(deviceLocks, i -> new Object());
+    }
 
     @Autowired
     private RestTemplate restTemplate;
 
     /** 离线判定超时，单位：分钟，默认30分钟 */
     @Value("${device.offline.timeout-minutes:30}")
-    private int offlineTimeoutMinutes;
+    private int offlineTimeoutMinutes = 30;
 
     /** 检查间隔，单位：毫秒，默认5分钟 */
     @Scheduled(fixedRateString = "${device.offline.check-interval-ms:300000}")
@@ -43,14 +51,16 @@ public class DeviceOfflineCheckTask {
         Date now = new Date();
         long timeoutMs = (long) offlineTimeoutMinutes * 60 * 1000;
         for (Map.Entry<String, Date> entry : deviceLastReceiveTimeMap.entrySet()) {
-            long diff = now.getTime() - entry.getValue().getTime();
-            if (diff > timeoutMs) {
-                // 只在设备不在离线集合中时才调用API
+            synchronized (deviceLock(entry.getKey())) {
+                // 与在线更新串行，重新读取时间，避免用旧快照覆盖刚收到的上报。
+                Date latest = deviceLastReceiveTimeMap.get(entry.getKey());
+                if (latest == null || now.getTime() - latest.getTime() <= timeoutMs) continue;
                 if (offlineDeviceSet.add(entry.getKey())) {
                     if (!updateDeviceOnlineStatus(entry.getKey(), 0)) {
                         offlineDeviceSet.remove(entry.getKey());
                         continue;
                     }
+                    onlineDeviceSet.remove(entry.getKey());
                     logger.info("设备 {} 已离线（超过{}分钟未收到数据）", entry.getKey(), offlineTimeoutMinutes);
                 }
             }
@@ -59,16 +69,24 @@ public class DeviceOfflineCheckTask {
 
     /**
      * 标记设备在线（收到数据时调用）
-     * 只在设备当前处于离线状态时才调用API
+     * 首次上报、服务重启或离线恢复均需同步；失败保留下次上报的重试机会。
      */
     public void markDeviceOnline(String deviceCode) {
-        if (deviceCode != null && offlineDeviceSet.contains(deviceCode)
-                && updateDeviceOnlineStatus(deviceCode, 1)) {
-            offlineDeviceSet.remove(deviceCode);
-            logger.info("设备 {} 恢复在线", deviceCode);
+        if (deviceCode == null || deviceCode.trim().isEmpty()) return;
+        synchronized (deviceLock(deviceCode)) {
+            // 仅在接口明确成功后缓存；新建默认离线的设备首次上报也会进入这里。
+            if (!onlineDeviceSet.contains(deviceCode) && updateDeviceOnlineStatus(deviceCode, 1)) {
+                onlineDeviceSet.add(deviceCode);
+                offlineDeviceSet.remove(deviceCode);
+                logger.info("设备 {} 已在线", deviceCode);
+            }
         }
     }
 
+    private Object deviceLock(String deviceCode) {
+        return deviceLocks[(deviceCode.hashCode() & Integer.MAX_VALUE) % deviceLocks.length];
+    }
+
     private boolean updateDeviceOnlineStatus(String deviceCode, int onlineStatus) {
         try {
             Map<String, Object> params = Map.of("deviceCode", deviceCode, "onlineStatus", onlineStatus);
@@ -86,4 +104,4 @@ public class DeviceOfflineCheckTask {
             return false;
         }
     }
-}
+}

firefighting-pressure-service/src/main/java/com/zksy/pressure/utils/DeviceOfflineCheckTask.java

@@ -12,6 +12,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestTemplate;
 
 import java.util.Date;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -29,13 +30,20 @@ public class DeviceOfflineCheckTask {
 
     /** 已知离线设备集合，用于避免重复API调用 */
     private final Set<String> offlineDeviceSet = ConcurrentHashMap.newKeySet();
+    /** 本实例已成功同步在线的设备；不能把尚未检测过离线当作已经在线。 */
+    private final Set<String> onlineDeviceSet = ConcurrentHashMap.newKeySet();
+    private final Object[] deviceLocks = new Object[128];
+
+    public DeviceOfflineCheckTask() {
+        Arrays.setAll(deviceLocks, i -> new Object());
+    }
 
     @Autowired
     private RestTemplate restTemplate;
 
     /** 离线判定超时，单位：分钟，默认30分钟 */
     @Value("${device.offline.timeout-minutes:30}")
-    private int offlineTimeoutMinutes;
+    private int offlineTimeoutMinutes = 30;
 
     /** 检查间隔，单位：毫秒，默认5分钟 */
     @Scheduled(fixedRateString = "${device.offline.check-interval-ms:300000}")
@@ -43,14 +51,16 @@ public class DeviceOfflineCheckTask {
         Date now = new Date();
         long timeoutMs = (long) offlineTimeoutMinutes * 60 * 1000;
         for (Map.Entry<String, Date> entry : deviceLastReceiveTimeMap.entrySet()) {
-            long diff = now.getTime() - entry.getValue().getTime();
-            if (diff > timeoutMs) {
-                // 只在设备不在离线集合中时才调用API
+            synchronized (deviceLock(entry.getKey())) {
+                // 与在线更新串行，重新读取时间，避免用旧快照覆盖刚收到的上报。
+                Date latest = deviceLastReceiveTimeMap.get(entry.getKey());
+                if (latest == null || now.getTime() - latest.getTime() <= timeoutMs) continue;
                 if (offlineDeviceSet.add(entry.getKey())) {
                     if (!updateDeviceOnlineStatus(entry.getKey(), 0)) {
                         offlineDeviceSet.remove(entry.getKey());
                         continue;
                     }
+                    onlineDeviceSet.remove(entry.getKey());
                     logger.info("设备 {} 已离线（超过{}分钟未收到数据）", entry.getKey(), offlineTimeoutMinutes);
                 }
             }
@@ -59,16 +69,24 @@ public class DeviceOfflineCheckTask {
 
     /**
      * 标记设备在线（收到数据时调用）
-     * 只在设备当前处于离线状态时才调用API
+     * 首次上报、服务重启或离线恢复均需同步；失败保留下次上报的重试机会。
      */
     public void markDeviceOnline(String deviceCode) {
-        if (deviceCode != null && offlineDeviceSet.contains(deviceCode)
-                && updateDeviceOnlineStatus(deviceCode, 1)) {
-            offlineDeviceSet.remove(deviceCode);
-            logger.info("设备 {} 恢复在线", deviceCode);
+        if (deviceCode == null || deviceCode.trim().isEmpty()) return;
+        synchronized (deviceLock(deviceCode)) {
+            // 仅在接口明确成功后缓存；新建默认离线的设备首次上报也会进入这里。
+            if (!onlineDeviceSet.contains(deviceCode) && updateDeviceOnlineStatus(deviceCode, 1)) {
+                onlineDeviceSet.add(deviceCode);
+                offlineDeviceSet.remove(deviceCode);
+                logger.info("设备 {} 已在线", deviceCode);
+            }
         }
     }
 
+    private Object deviceLock(String deviceCode) {
+        return deviceLocks[(deviceCode.hashCode() & Integer.MAX_VALUE) % deviceLocks.length];
+    }
+
     private boolean updateDeviceOnlineStatus(String deviceCode, int onlineStatus) {
         try {
             Map<String, Object> params = Map.of("deviceCode", deviceCode, "onlineStatus", onlineStatus);
@@ -86,4 +104,4 @@ public class DeviceOfflineCheckTask {
             return false;
         }
     }
-}
+}

flammable-gas-service/src/main/java/com/zksy/gas/utils/DeviceOfflineCheckTask.java

@@ -12,6 +12,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestTemplate;
 
 import java.util.Date;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -23,25 +24,35 @@ public class DeviceOfflineCheckTask {
     public static ConcurrentHashMap<String, Date> deviceLastReceiveTimeMap = new ConcurrentHashMap<>();
 
     private final Set<String> offlineDeviceSet = ConcurrentHashMap.newKeySet();
+    /** 本实例已成功同步在线的设备；不能把尚未检测过离线当作已经在线。 */
+    private final Set<String> onlineDeviceSet = ConcurrentHashMap.newKeySet();
+    private final Object[] deviceLocks = new Object[128];
+
+    public DeviceOfflineCheckTask() {
+        Arrays.setAll(deviceLocks, i -> new Object());
+    }
 
     @Autowired
     private RestTemplate restTemplate;
 
     @Value("${device.offline.timeout-minutes:30}")
-    private int offlineTimeoutMinutes;
+    private int offlineTimeoutMinutes = 30;
 
     @Scheduled(fixedRateString = "${device.offline.check-interval-ms:300000}")
     public void checkDeviceOffline() {
         Date now = new Date();
         long timeoutMs = (long) offlineTimeoutMinutes * 60 * 1000;
         for (Map.Entry<String, Date> entry : deviceLastReceiveTimeMap.entrySet()) {
-            long diff = now.getTime() - entry.getValue().getTime();
-            if (diff > timeoutMs) {
+            synchronized (deviceLock(entry.getKey())) {
+                // 与在线更新串行，重新读取时间，避免用旧快照覆盖刚收到的上报。
+                Date latest = deviceLastReceiveTimeMap.get(entry.getKey());
+                if (latest == null || now.getTime() - latest.getTime() <= timeoutMs) continue;
                 if (offlineDeviceSet.add(entry.getKey())) {
                     if (!updateDeviceOnlineStatus(entry.getKey(), 0)) {
                         offlineDeviceSet.remove(entry.getKey());
                         continue;
                     }
+                    onlineDeviceSet.remove(entry.getKey());
                     logger.info("设备 {} 已离线（超过{}分钟未收到数据）", entry.getKey(), offlineTimeoutMinutes);
                 }
             }
@@ -49,13 +60,21 @@ public class DeviceOfflineCheckTask {
     }
 
     public void markDeviceOnline(String deviceCode) {
-        if (deviceCode != null && offlineDeviceSet.contains(deviceCode)
-                && updateDeviceOnlineStatus(deviceCode, 1)) {
-            offlineDeviceSet.remove(deviceCode);
-            logger.info("设备 {} 恢复在线", deviceCode);
+        if (deviceCode == null || deviceCode.trim().isEmpty()) return;
+        synchronized (deviceLock(deviceCode)) {
+            // 仅在接口明确成功后缓存；新建默认离线的设备首次上报也会进入这里。
+            if (!onlineDeviceSet.contains(deviceCode) && updateDeviceOnlineStatus(deviceCode, 1)) {
+                onlineDeviceSet.add(deviceCode);
+                offlineDeviceSet.remove(deviceCode);
+                logger.info("设备 {} 已在线", deviceCode);
+            }
         }
     }
 
+    private Object deviceLock(String deviceCode) {
+        return deviceLocks[(deviceCode.hashCode() & Integer.MAX_VALUE) % deviceLocks.length];
+    }
+
     private boolean updateDeviceOnlineStatus(String deviceCode, int onlineStatus) {
         try {
             Map<String, Object> params = Map.of("deviceCode", deviceCode, "onlineStatus", onlineStatus);
@@ -73,4 +92,4 @@ public class DeviceOfflineCheckTask {
             return false;
         }
     }
-}
+}

manhole-service/src/main/java/com/zksy/manhole/config/RestTemplateConfig.java

@@ -1,6 +1,7 @@
 package com.zksy.manhole.config;
 
 import org.springframework.context.annotation.Bean;
+import org.springframework.cloud.client.loadbalancer.LoadBalanced;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.web.client.RestTemplate;
 import org.springframework.beans.factory.annotation.Value;
@@ -10,6 +11,7 @@ import org.springframework.http.client.SimpleClientHttpRequestFactory;
 public class RestTemplateConfig {
 
     @Bean
+    @LoadBalanced
     public RestTemplate restTemplate(
             @Value("${collection.http.connect-timeout-ms:3000}") int connectTimeout,
             @Value("${collection.http.read-timeout-ms:5000}") int readTimeout) {
@@ -18,4 +20,4 @@ public class RestTemplateConfig {
         factory.setReadTimeout(Math.max(1, readTimeout));
         return new RestTemplate(factory);
     }
-}
+}

manhole-service/src/main/java/com/zksy/manhole/utils/DeviceOfflineCheckTask.java

@@ -12,6 +12,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestTemplate;
 
 import java.util.Date;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -29,13 +30,20 @@ public class DeviceOfflineCheckTask {
 
     /** 已知离线设备集合，用于避免重复API调用 */
     private final Set<String> offlineDeviceSet = ConcurrentHashMap.newKeySet();
+    /** 本实例已成功同步在线的设备；不能把尚未检测过离线当作已经在线。 */
+    private final Set<String> onlineDeviceSet = ConcurrentHashMap.newKeySet();
+    private final Object[] deviceLocks = new Object[128];
+
+    public DeviceOfflineCheckTask() {
+        Arrays.setAll(deviceLocks, i -> new Object());
+    }
 
     @Autowired
     private RestTemplate restTemplate;
 
     /** 离线判定超时，单位：分钟，默认30分钟 */
     @Value("${device.offline.timeout-minutes:30}")
-    private int offlineTimeoutMinutes;
+    private int offlineTimeoutMinutes = 30;
 
     /** 检查间隔，单位：毫秒，默认5分钟 */
     @Scheduled(fixedRateString = "${device.offline.check-interval-ms:300000}")
@@ -43,14 +51,16 @@ public class DeviceOfflineCheckTask {
         Date now = new Date();
         long timeoutMs = (long) offlineTimeoutMinutes * 60 * 1000;
         for (Map.Entry<String, Date> entry : deviceLastReceiveTimeMap.entrySet()) {
-            long diff = now.getTime() - entry.getValue().getTime();
-            if (diff > timeoutMs) {
-                // 只在设备不在离线集合中时才调用API
+            synchronized (deviceLock(entry.getKey())) {
+                // 与在线更新串行，重新读取时间，避免用旧快照覆盖刚收到的上报。
+                Date latest = deviceLastReceiveTimeMap.get(entry.getKey());
+                if (latest == null || now.getTime() - latest.getTime() <= timeoutMs) continue;
                 if (offlineDeviceSet.add(entry.getKey())) {
                     if (!updateDeviceOnlineStatus(entry.getKey(), 0)) {
                         offlineDeviceSet.remove(entry.getKey());
                         continue;
                     }
+                    onlineDeviceSet.remove(entry.getKey());
                     logger.info("设备 {} 已离线（超过{}分钟未收到数据）", entry.getKey(), offlineTimeoutMinutes);
                 }
             }
@@ -59,16 +69,24 @@ public class DeviceOfflineCheckTask {
 
     /**
      * 标记设备在线（收到数据时调用）
-     * 只在设备当前处于离线状态时才调用API
+     * 首次上报、服务重启或离线恢复均需同步；失败保留下次上报的重试机会。
      */
     public void markDeviceOnline(String deviceCode) {
-        if (deviceCode != null && offlineDeviceSet.contains(deviceCode)
-                && updateDeviceOnlineStatus(deviceCode, 1)) {
-            offlineDeviceSet.remove(deviceCode);
-            logger.info("设备 {} 恢复在线", deviceCode);
+        if (deviceCode == null || deviceCode.trim().isEmpty()) return;
+        synchronized (deviceLock(deviceCode)) {
+            // 仅在接口明确成功后缓存；新建默认离线的设备首次上报也会进入这里。
+            if (!onlineDeviceSet.contains(deviceCode) && updateDeviceOnlineStatus(deviceCode, 1)) {
+                onlineDeviceSet.add(deviceCode);
+                offlineDeviceSet.remove(deviceCode);
+                logger.info("设备 {} 已在线", deviceCode);
+            }
         }
     }
 
+    private Object deviceLock(String deviceCode) {
+        return deviceLocks[(deviceCode.hashCode() & Integer.MAX_VALUE) % deviceLocks.length];
+    }
+
     private boolean updateDeviceOnlineStatus(String deviceCode, int onlineStatus) {
         try {
             Map<String, Object> params = Map.of("deviceCode", deviceCode, "onlineStatus", onlineStatus);
@@ -86,4 +104,4 @@ public class DeviceOfflineCheckTask {
             return false;
         }
     }
-}
+}

radar-service/src/main/java/com/zksy/radar/utils/DeviceOfflineCheckTask.java

@@ -12,6 +12,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestTemplate;
 
 import java.util.Date;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -23,25 +24,35 @@ public class DeviceOfflineCheckTask {
     public static ConcurrentHashMap<String, Date> deviceLastReceiveTimeMap = new ConcurrentHashMap<>();
 
     private final Set<String> offlineDeviceSet = ConcurrentHashMap.newKeySet();
+    /** 本实例已成功同步在线的设备；不能把尚未检测过离线当作已经在线。 */
+    private final Set<String> onlineDeviceSet = ConcurrentHashMap.newKeySet();
+    private final Object[] deviceLocks = new Object[128];
+
+    public DeviceOfflineCheckTask() {
+        Arrays.setAll(deviceLocks, i -> new Object());
+    }
 
     @Autowired
     private RestTemplate restTemplate;
 
     @Value("${device.offline.timeout-minutes:30}")
-    private int offlineTimeoutMinutes;
+    private int offlineTimeoutMinutes = 30;
 
     @Scheduled(fixedRateString = "${device.offline.check-interval-ms:300000}")
     public void checkDeviceOffline() {
         Date now = new Date();
         long timeoutMs = (long) offlineTimeoutMinutes * 60 * 1000;
         for (Map.Entry<String, Date> entry : deviceLastReceiveTimeMap.entrySet()) {
-            long diff = now.getTime() - entry.getValue().getTime();
-            if (diff > timeoutMs) {
+            synchronized (deviceLock(entry.getKey())) {
+                // 与在线更新串行，重新读取时间，避免用旧快照覆盖刚收到的上报。
+                Date latest = deviceLastReceiveTimeMap.get(entry.getKey());
+                if (latest == null || now.getTime() - latest.getTime() <= timeoutMs) continue;
                 if (offlineDeviceSet.add(entry.getKey())) {
                     if (!updateDeviceOnlineStatus(entry.getKey(), 0)) {
                         offlineDeviceSet.remove(entry.getKey());
                         continue;
                     }
+                    onlineDeviceSet.remove(entry.getKey());
                     logger.info("设备 {} 已离线（超过{}分钟未收到数据）", entry.getKey(), offlineTimeoutMinutes);
                 }
             }
@@ -49,13 +60,21 @@ public class DeviceOfflineCheckTask {
     }
 
     public void markDeviceOnline(String deviceCode) {
-        if (deviceCode != null && offlineDeviceSet.contains(deviceCode)
-                && updateDeviceOnlineStatus(deviceCode, 1)) {
-            offlineDeviceSet.remove(deviceCode);
-            logger.info("设备 {} 恢复在线", deviceCode);
+        if (deviceCode == null || deviceCode.trim().isEmpty()) return;
+        synchronized (deviceLock(deviceCode)) {
+            // 仅在接口明确成功后缓存；新建默认离线的设备首次上报也会进入这里。
+            if (!onlineDeviceSet.contains(deviceCode) && updateDeviceOnlineStatus(deviceCode, 1)) {
+                onlineDeviceSet.add(deviceCode);
+                offlineDeviceSet.remove(deviceCode);
+                logger.info("设备 {} 已在线", deviceCode);
+            }
         }
     }
 
+    private Object deviceLock(String deviceCode) {
+        return deviceLocks[(deviceCode.hashCode() & Integer.MAX_VALUE) % deviceLocks.length];
+    }
+
     private boolean updateDeviceOnlineStatus(String deviceCode, int onlineStatus) {
         try {
             Map<String, Object> params = Map.of("deviceCode", deviceCode, "onlineStatus", onlineStatus);
@@ -73,4 +92,4 @@ public class DeviceOfflineCheckTask {
             return false;
         }
     }
-}
+}

telemetry-service/src/main/java/com/zksy/telemetry/utils/DeviceOfflineCheckTask.java

@@ -12,6 +12,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestTemplate;
 
 import java.util.Date;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -23,25 +24,35 @@ public class DeviceOfflineCheckTask {
     public static ConcurrentHashMap<String, Date> deviceLastReceiveTimeMap = new ConcurrentHashMap<>();
 
     private final Set<String> offlineDeviceSet = ConcurrentHashMap.newKeySet();
+    /** 本实例已成功同步在线的设备；不能把尚未检测过离线当作已经在线。 */
+    private final Set<String> onlineDeviceSet = ConcurrentHashMap.newKeySet();
+    private final Object[] deviceLocks = new Object[128];
+
+    public DeviceOfflineCheckTask() {
+        Arrays.setAll(deviceLocks, i -> new Object());
+    }
 
     @Autowired
     private RestTemplate restTemplate;
 
     @Value("${device.offline.timeout-minutes:30}")
-    private int offlineTimeoutMinutes;
+    private int offlineTimeoutMinutes = 30;
 
     @Scheduled(fixedRateString = "${device.offline.check-interval-ms:300000}")
     public void checkDeviceOffline() {
         Date now = new Date();
         long timeoutMs = (long) offlineTimeoutMinutes * 60 * 1000;
         for (Map.Entry<String, Date> entry : deviceLastReceiveTimeMap.entrySet()) {
-            long diff = now.getTime() - entry.getValue().getTime();
-            if (diff > timeoutMs) {
+            synchronized (deviceLock(entry.getKey())) {
+                // 与在线更新串行，重新读取时间，避免用旧快照覆盖刚收到的上报。
+                Date latest = deviceLastReceiveTimeMap.get(entry.getKey());
+                if (latest == null || now.getTime() - latest.getTime() <= timeoutMs) continue;
                 if (offlineDeviceSet.add(entry.getKey())) {
                     if (!updateDeviceOnlineStatus(entry.getKey(), 0)) {
                         offlineDeviceSet.remove(entry.getKey());
                         continue;
                     }
+                    onlineDeviceSet.remove(entry.getKey());
                     logger.info("设备 {} 已离线（超过{}分钟未收到数据）", entry.getKey(), offlineTimeoutMinutes);
                 }
             }
@@ -49,13 +60,21 @@ public class DeviceOfflineCheckTask {
     }
 
     public void markDeviceOnline(String deviceCode) {
-        if (deviceCode != null && offlineDeviceSet.contains(deviceCode)
-                && updateDeviceOnlineStatus(deviceCode, 1)) {
-            offlineDeviceSet.remove(deviceCode);
-            logger.info("设备 {} 恢复在线", deviceCode);
+        if (deviceCode == null || deviceCode.trim().isEmpty()) return;
+        synchronized (deviceLock(deviceCode)) {
+            // 仅在接口明确成功后缓存；新建默认离线的设备首次上报也会进入这里。
+            if (!onlineDeviceSet.contains(deviceCode) && updateDeviceOnlineStatus(deviceCode, 1)) {
+                onlineDeviceSet.add(deviceCode);
+                offlineDeviceSet.remove(deviceCode);
+                logger.info("设备 {} 已在线", deviceCode);
+            }
         }
     }
 
+    private Object deviceLock(String deviceCode) {
+        return deviceLocks[(deviceCode.hashCode() & Integer.MAX_VALUE) % deviceLocks.length];
+    }
+
     private boolean updateDeviceOnlineStatus(String deviceCode, int onlineStatus) {
         try {
             Map<String, Object> params = Map.of("deviceCode", deviceCode, "onlineStatus", onlineStatus);
@@ -73,4 +92,4 @@ public class DeviceOfflineCheckTask {
             return false;
         }
     }
-}
+}

water-level-service/src/main/java/com/zksy/water/utils/DeviceOfflineCheckTask.java

@@ -12,6 +12,7 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestTemplate;
 
 import java.util.Date;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -23,25 +24,35 @@ public class DeviceOfflineCheckTask {
     public static ConcurrentHashMap<String, Date> deviceLastReceiveTimeMap = new ConcurrentHashMap<>();
 
     private final Set<String> offlineDeviceSet = ConcurrentHashMap.newKeySet();
+    /** 本实例已成功同步在线的设备；不能把尚未检测过离线当作已经在线。 */
+    private final Set<String> onlineDeviceSet = ConcurrentHashMap.newKeySet();
+    private final Object[] deviceLocks = new Object[128];
+
+    public DeviceOfflineCheckTask() {
+        Arrays.setAll(deviceLocks, i -> new Object());
+    }
 
     @Autowired
     private RestTemplate restTemplate;
 
     @Value("${device.offline.timeout-minutes:30}")
-    private int offlineTimeoutMinutes;
+    private int offlineTimeoutMinutes = 30;
 
     @Scheduled(fixedRateString = "${device.offline.check-interval-ms:300000}")
     public void checkDeviceOffline() {
         Date now = new Date();
         long timeoutMs = (long) offlineTimeoutMinutes * 60 * 1000;
         for (Map.Entry<String, Date> entry : deviceLastReceiveTimeMap.entrySet()) {
-            long diff = now.getTime() - entry.getValue().getTime();
-            if (diff > timeoutMs) {
+            synchronized (deviceLock(entry.getKey())) {
+                // 与在线更新串行，重新读取时间，避免用旧快照覆盖刚收到的上报。
+                Date latest = deviceLastReceiveTimeMap.get(entry.getKey());
+                if (latest == null || now.getTime() - latest.getTime() <= timeoutMs) continue;
                 if (offlineDeviceSet.add(entry.getKey())) {
                     if (!updateDeviceOnlineStatus(entry.getKey(), 0)) {
                         offlineDeviceSet.remove(entry.getKey());
                         continue;
                     }
+                    onlineDeviceSet.remove(entry.getKey());
                     logger.info("设备 {} 已离线（超过{}分钟未收到数据）", entry.getKey(), offlineTimeoutMinutes);
                 }
             }
@@ -49,13 +60,21 @@ public class DeviceOfflineCheckTask {
     }
 
     public void markDeviceOnline(String deviceCode) {
-        if (deviceCode != null && offlineDeviceSet.contains(deviceCode)
-                && updateDeviceOnlineStatus(deviceCode, 1)) {
-            offlineDeviceSet.remove(deviceCode);
-            logger.info("设备 {} 恢复在线", deviceCode);
+        if (deviceCode == null || deviceCode.trim().isEmpty()) return;
+        synchronized (deviceLock(deviceCode)) {
+            // 仅在接口明确成功后缓存；新建默认离线的设备首次上报也会进入这里。
+            if (!onlineDeviceSet.contains(deviceCode) && updateDeviceOnlineStatus(deviceCode, 1)) {
+                onlineDeviceSet.add(deviceCode);
+                offlineDeviceSet.remove(deviceCode);
+                logger.info("设备 {} 已在线", deviceCode);
+            }
         }
     }
 
+    private Object deviceLock(String deviceCode) {
+        return deviceLocks[(deviceCode.hashCode() & Integer.MAX_VALUE) % deviceLocks.length];
+    }
+
     private boolean updateDeviceOnlineStatus(String deviceCode, int onlineStatus) {
         try {
             Map<String, Object> params = Map.of("deviceCode", deviceCode, "onlineStatus", onlineStatus);
@@ -73,4 +92,4 @@ public class DeviceOfflineCheckTask {
             return false;
         }
     }
-}
+}