fix(websocket): 使用redis解决多节点问题

src/main/java/com/dlbridge/jadp/config/web/RedisConfig.java

 package com.dlbridge.jadp.config.web;
 
+import com.dlbridge.jadp.websocket.WebSocketRedisSubscriber;
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.PropertyAccessor;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.extern.log4j.Log4j;
-import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Profile;
@@ -13,14 +13,14 @@ import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.RedisNode;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
 import org.springframework.data.redis.core.*;
+import org.springframework.data.redis.listener.ChannelTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
 import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 import org.springframework.stereotype.Component;
 import redis.clients.jedis.JedisPoolConfig;
 
-import javax.annotation.Resource;
-import java.util.List;
-
 /**
  * date  : 2019/1/21
  * author : yangshuxing
@@ -40,6 +40,27 @@ public class RedisConfig {
         return config;
     }
 
+    @Bean
+    public ChannelTopic topic() {
+        return new ChannelTopic("websocket:message");
+    }
+
+    @Bean
+    public RedisMessageListenerContainer redisContainer(
+            RedisConnectionFactory connectionFactory,
+            MessageListenerAdapter listenerAdapter,
+            ChannelTopic topic) {
+        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+        container.setConnectionFactory(connectionFactory);
+        container.addMessageListener(listenerAdapter, topic);
+        return container;
+    }
+
+    @Bean
+    public MessageListenerAdapter listenerAdapter(WebSocketRedisSubscriber subscriber) {
+        return new MessageListenerAdapter(subscriber, "onMessage");
+    }
+
     @Bean
     @Profile("production")
     public RedisClusterConfiguration predisClusterConfiguration() {
@@ -169,4 +190,6 @@ public class RedisConfig {
         return redisTemplate.opsForZSet();
     }
 
+
+
 }

src/main/java/com/dlbridge/jadp/websocket/WebSocketMessage.java

@@ -53,7 +53,7 @@ public class WebSocketMessage {
         return new WebSocketMessage(MessageType.INFO.getValue(), ticketNo, message, data);
     }
 
-    // 转 JSON 字符串
+    // ✅ 转 JSON 字符串
     public String toJson() {
         JSONObject obj = JSONUtil.createObj()
                 .set("type", type)
@@ -66,4 +66,19 @@ public class WebSocketMessage {
 
         return obj.toString();
     }
+
+    // ✅ 从 JSON 字符串反序列化（自动修复多重转义问题）
+    public static WebSocketMessage fromJson(String jsonStr) {
+        if (jsonStr == null) {
+            return null;
+        }
+
+        // 处理双重转义的情况：比如 "{\"type\":\"success\"...}"
+        if (jsonStr.startsWith("\"") && jsonStr.endsWith("\"")) {
+            jsonStr = jsonStr.substring(1, jsonStr.length() - 1);
+            jsonStr = jsonStr.replace("\\\"", "\"");
+        }
+
+        return JSONUtil.toBean(jsonStr, WebSocketMessage.class);
+    }
 }
\ No newline at end of file

src/main/java/com/dlbridge/jadp/websocket/WebSocketRedisSubscriber.java

+package com.dlbridge.jadp.websocket;
+import cn.hutool.json.JSONUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+@Slf4j
+@Component
+public class WebSocketRedisSubscriber implements MessageListener {
+
+    private final WebSocketSessionService sessionService;
+
+    public WebSocketRedisSubscriber(WebSocketSessionService sessionService) {
+        this.sessionService = sessionService;
+    }
+
+    @Override
+    public void onMessage(Message message, byte[] pattern) {
+        try {
+            String msgStr = new String(message.getBody());
+            log.info("接收到 Redis 分发的 WebSocket 消息: {}", msgStr);
+            // ✅ 自动修复双引号问题
+            WebSocketMessage wsMsg = WebSocketMessage.fromJson(msgStr);
+            sessionService.sendToUser(wsMsg.getTicketNo(), wsMsg);
+        } catch (Exception e) {
+            log.error("Redis 消息处理异常: {}", e.getMessage(), e);
+        }
+    }
+}

src/main/java/com/dlbridge/jadp/websocket/WebSocketSessionService.java

 package com.dlbridge.jadp.websocket;
 
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.ChannelTopic;
 import org.springframework.stereotype.Service;
 
-import java.util.*;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 @Slf4j
 @Service
 public class WebSocketSessionService {
 
+
+    private final RedisTemplate<String, Object> redisTemplate;
+    private final ChannelTopic topic;
+
+    public WebSocketSessionService(RedisTemplate<String, Object> redisTemplate, ChannelTopic topic) {
+        this.redisTemplate = redisTemplate;
+        this.topic = topic;
+    }
+
     private final Map<String, String> sessionUserMap = new ConcurrentHashMap<>();
     private final Map<String, Set<String>> userTicketsMap = new ConcurrentHashMap<>();
     private final Map<String, String> ticketUserMap = new ConcurrentHashMap<>();
@@ -68,7 +81,7 @@ public class WebSocketSessionService {
         });
         return sessionIds;
     }
-
+    // 发送消息给客户
     public void sendToUser(String userId, WebSocketMessage message) {
         String payload = message.toJson();
         getSessionIdsByUserId(userId).forEach(sessionId -> {
@@ -77,13 +90,24 @@ public class WebSocketSessionService {
         });
         log.info("用户通知已发送 - 用户: {}, 类型: {}", userId, message.getType());
     }
-
+    // 发送核销验证成功的通知
     public void sendVerificationSuccess(String ticketNo, String msg) {
+        WebSocketMessage message = WebSocketMessage.success(ticketNo, msg);
+        String userId = ticketUserMap.get(ticketNo);
+        // 本地推送
+        sendLocal(ticketNo, message);
+        // if (userId == null) {
+            // 发布到 Redis，让其他节点同步
+            redisTemplate.convertAndSend(topic.getTopic(), message.toJson());
+        // }
+    }
+    // 本地推送消息给客户
+    private void sendLocal(String ticketNo, WebSocketMessage message) {
         String userId = ticketUserMap.get(ticketNo);
-        if (userId != null) sendToUser(userId, WebSocketMessage.success(ticketNo, msg));
+        if (userId != null) sendToUser(userId, message);
         else log.warn("未找到用户，无法发送核销消息 - 票号: {}", ticketNo);
     }
-
+    // 发送错误消息给客户
     public void sendError(String ticketNo, String errorMsg) {
         String userId = ticketUserMap.get(ticketNo);
         if (userId != null) sendToUser(userId, WebSocketMessage.error(ticketNo, errorMsg));