流量补全消费者改为缓存排序处理，保证顺序。

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

@@ -28,7 +28,6 @@ import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClients;
 import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
 import org.apache.http.util.EntityUtils;
-import org.redisson.api.RLock;
 import org.redisson.api.RedissonClient;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -90,7 +89,8 @@ public class MessageHandler {
     private static final String USED_PORTS_KEY = "frpc:used_ports";
     private static final long HEARTBEAT_TIMEOUT = 30000; // 3分钟超时
 
-
+    private final ConcurrentHashMap<String, LinkedBlockingQueue<DeviceMessage>> messageBuffer =
+            new ConcurrentHashMap<>();
     @Autowired
     private RedisTemplate<String, String> redisTemplate;
     @Autowired
@@ -875,36 +875,59 @@ public class MessageHandler {
     }
     private void handleNetRecoverMessage(DeviceMessage message) {
         String clientId = message.getClientId();
-        String lockKey = "traffic:recover:" + clientId;
+        List<InitialBandwidthTraffic> interfaces = JsonDataParser.parseJsonData(message.getData(), InitialBandwidthTraffic.class);
-        RLock lock = redissonClient.getLock(lockKey);
-        boolean locked = false;
 
-        try {
+        if (interfaces.isEmpty()) {
-            // 尝试获取锁
+            return;
-            locked = lock.tryLock(0, 20, TimeUnit.SECONDS);
+        }
-            if (locked) {
+
-//                log.info("设备{}获取锁成功，开始处理消息", clientId);
+        boolean isLast = interfaces.get(0).isLastTrafficFlag();
-                processNetRecoverMessageInternal(message);
+
-//                log.info("设备{}消息处理完成", clientId);
+        // 1. 消息入队
-            } else {
+        LinkedBlockingQueue<DeviceMessage> queue = messageBuffer
-                log.warn("设备{}获取锁失败，消息处理繁忙", clientId);
+                .computeIfAbsent(clientId, k -> new LinkedBlockingQueue<>());
-                throw new RuntimeException("设备处理繁忙，请重试");
+        queue.offer(message);
-            }
+
-        } catch (InterruptedException e) {
+        // 2. 如果是最后一条消息，开始处理（此时前面的消息肯定都已经到了）
-            Thread.currentThread().interrupt();
+        if (isLast) {
-            log.error("处理消息时线程被中断", e);
+            processAllMessagesInOrder(clientId);
-        } finally {
-            // 只有成功获取锁的线程才需要释放
-            if (locked && lock.isHeldByCurrentThread()) {
-                try {
-                    lock.unlock();
-//                    log.debug("设备{}锁已释放", clientId);
-                } catch (IllegalMonitorStateException e) {
-                    log.warn("设备{}锁释放异常，可能已自动超时", clientId);
-                }
-            }
         }
     }
+
+    private void processAllMessagesInOrder(String clientId) {
+        LinkedBlockingQueue<DeviceMessage> queue = messageBuffer.get(clientId);
+        if (queue == null || queue.isEmpty()) {
+            return;
+        }
+
+        // 1. 取出所有消息
+        List<DeviceMessage> allMessages = new ArrayList<>();
+        queue.drainTo(allMessages);
+
+        // 2. 按时间戳排序
+        allMessages.sort((msg1, msg2) -> {
+            long ts1 = JsonDataParser.parseJsonData(msg1.getData(), InitialBandwidthTraffic.class)
+                    .get(0).getTimestamp();
+            long ts2 = JsonDataParser.parseJsonData(msg2.getData(), InitialBandwidthTraffic.class)
+                    .get(0).getTimestamp();
+            return Long.compare(ts1, ts2);
+        });
+
+        // 3. 顺序处理
+        for (DeviceMessage message : allMessages) {
+            try {
+                processNetRecoverMessageInternal(message);
+            } catch (Exception e) {
+                log.error("处理设备{}消息失败", clientId, e);
+            }
+        }
+
+        log.info("设备{}顺序处理完成，共处理{}条消息", clientId, allMessages.size());
+
+        // 4. 清理
+        messageBuffer.remove(clientId);
+    }
+
     /**
      * 网络重试流量数据入库
      * @param message