feat(ws): 实现WebSocket聊天功能增强与优化

- 新增RedisTemplate支持，用于存储用户客服分配状态
- 优化用户和客服上下线通知机制
- 增加消息已读回执处理逻辑
- 添加用户输入状态（正在输入）转发功能
- 实现客服分配提醒间隔控制，避免频繁提示
- 完善心跳、登出等消息类型的处理
- 改进连接超时处理逻辑，统一线程管理
- 升级Netty配置，启用WebSocket协议扩展支持
- 调整Java版本至11，更新相关依赖配置
- 新增zksy-ws模块，重构WebSocket服务结构

pom.xml

@@ -2,8 +2,8 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0"
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-	<modelVersion>4.0.0</modelVersion>
-	
+    <modelVersion>4.0.0</modelVersion>
+
     <groupId>com.zksy</groupId>
     <artifactId>zksy</artifactId>
     <version>3.9.0</version>
@@ -11,15 +11,15 @@
     <name>zksy</name>
     <url>http://www.ruoyi.vip</url>
     <description>中科盛阳官网管理系统</description>
-    
+
     <properties>
         <zksy.version>3.9.0</zksy.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
-        <java.version>1.8</java.version>
+        <java.version>11</java.version>
         <maven-jar-plugin.version>3.1.1</maven-jar-plugin.version>
         <spring-boot.version>2.5.15</spring-boot.version>
-        <mybatis-plus.version>3.5.6</mybatis-plus.version>
+        <mybatis-plus.version>3.5.3.1</mybatis-plus.version>
         <org.projectlombok.version>1.18.20</org.projectlombok.version>
         <druid.version>1.2.23</druid.version>
         <bitwalker.version>1.21</bitwalker.version>
@@ -228,6 +228,11 @@
                 <artifactId>zksy-common</artifactId>
                 <version>${zksy.version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.zksy</groupId>
+                <artifactId>zksy-ws</artifactId>
+                <version>${zksy.version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.github.xiaoymin</groupId>
                 <artifactId>knife4j-openapi2-spring-boot-starter</artifactId>
@@ -243,6 +248,7 @@
         <module>zksy-quartz</module>
         <module>zksy-generator</module>
         <module>zksy-common</module>
+        <module>zksy-ws</module>
     </modules>
     <packaging>pom</packaging>
 

zksy-admin/pom.xml

@@ -72,6 +72,11 @@
             <artifactId>zksy-generator</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>com.zksy</groupId>
+            <artifactId>zksy-system</artifactId>
+        </dependency>
+
     </dependencies>
 
     <build>

zksy-admin/src/main/java/com/zksy/ZksyApplication.java

@@ -7,6 +7,9 @@ import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
+import org.springframework.data.redis.core.RedisTemplate;
+
+import javax.annotation.Resource;
 
 /**
  * 启动程序
@@ -18,6 +21,9 @@ public class ZksyApplication implements CommandLineRunner {
 
     @Autowired
     private ChatService chatService;
+
+    @Resource
+    private RedisTemplate<Object,Object> redisTemplate;
     public static void main(String[] args) {
         // System.setProperty("spring.devtools.restart.enabled", "false");
         SpringApplication.run(ZksyApplication.class, args);
@@ -28,7 +34,7 @@ public class ZksyApplication implements CommandLineRunner {
     public void run(String... args) throws Exception {
         new Thread(() -> {
             try {
-                new WebSocketServer(8081, chatService).start();
+                new WebSocketServer(8081, chatService,redisTemplate).start();
             } catch (InterruptedException e) {
                 e.printStackTrace();
             }

zksy-admin/src/main/java/com/zksy/web/controller/base/NewsUpdatesController.java

@@ -44,7 +44,9 @@ public class NewsUpdatesController {
     public AjaxResult findByPage(@ApiParam(value = "页码", required = true)long pageNum,
                                  @ApiParam(value = "页数", required = true)long pageSize){
         Page<NewsUpdates> page = new Page<>(pageNum, pageSize);
-        return AjaxResult.success(service.page(page));
+        LambdaQueryWrapper<NewsUpdates> queryWrapper = new LambdaQueryWrapper<>();
+        queryWrapper.orderByDesc(NewsUpdates::getCreateTime);
+        return AjaxResult.success(service.page(page,queryWrapper));
     }
 
     @GetMapping("/getNewsUpdatesList")

zksy-common/pom.xml

@@ -28,6 +28,11 @@
             <groupId>org.springframework</groupId>
             <artifactId>spring-web</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+            <version>4.1.108.Final</version>
+        </dependency>
 
         <!-- spring security 安全认证 -->
         <dependency>
@@ -41,6 +46,18 @@
             <artifactId>pagehelper-spring-boot-starter</artifactId>
         </dependency>
 
+        <!-- 引入MyBatis-Plus依赖 -->
+        <dependency>
+            <groupId>com.baomidou</groupId>
+            <artifactId>mybatis-plus-boot-starter</artifactId>
+        </dependency>
+        <!-- 添加Lombok依赖 -->
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+
+
         <!-- 自定义验证注解 -->
         <dependency>
             <groupId>org.springframework.boot</groupId>

zksy-system/pom.xml

@@ -13,8 +13,8 @@
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
                 <configuration>
-                    <source>9</source>
-                    <target>9</target>
+                    <source>11</source>
+                    <target>11</target>
                 </configuration>
             </plugin>
         </plugins>
@@ -34,15 +34,10 @@
             <groupId>com.zksy</groupId>
             <artifactId>zksy-common</artifactId>
         </dependency>
-        <!-- 引入MyBatis-Plus依赖 -->
-        <dependency>
-            <groupId>com.baomidou</groupId>
-            <artifactId>mybatis-plus-boot-starter</artifactId>
-        </dependency>
-        <!-- 添加Lombok依赖 -->
+
         <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
+            <groupId>com.zksy</groupId>
+            <artifactId>zksy-ws</artifactId>
         </dependency>
     </dependencies>
 

zksy-ws/pom.xml

@@ -0,0 +1,19 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>com.zksy</groupId>
+        <artifactId>zksy</artifactId>
+        <version>3.9.0</version>
+    </parent>
+    <artifactId>zksy-ws</artifactId>
+    <name>zksy-ws</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.zksy</groupId>
+            <artifactId>zksy-common</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

zksy-ws/src/main/java/com/zksy/server/WebSocketServer.java

@@ -6,15 +6,14 @@ import io.netty.channel.ChannelFuture;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
+import org.springframework.data.redis.core.RedisTemplate;
+import lombok.RequiredArgsConstructor;
 
+@RequiredArgsConstructor
 public class WebSocketServer {
     private final int port;
     private final ChatService chatService;
-
-    public WebSocketServer(int port, ChatService chatService) {
-        this.port = port;
-        this.chatService = chatService;
-    }
+    private final RedisTemplate<Object,Object> redisTemplate;
 
     public void start() throws InterruptedException {
         EventLoopGroup boss = new NioEventLoopGroup(1);
@@ -24,7 +23,7 @@ public class WebSocketServer {
             ServerBootstrap bootstrap = new ServerBootstrap();
             bootstrap.group(boss, worker)
                      .channel(NioServerSocketChannel.class)
-                     .childHandler(new WebSocketServerInitializer(chatService));
+                     .childHandler(new WebSocketServerInitializer(chatService,redisTemplate));
 
             ChannelFuture future = bootstrap.bind(port).sync();
             System.out.println("WebSocket 服务器已启动，端口：" + port);

zksy-ws/src/main/java/com/zksy/server/WebSocketServerHandler.java

@@ -1,5 +1,6 @@
 package com.zksy.server;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.zksy.domain.dto.ChatMessageDto;
 import com.zksy.service.ChatService;
@@ -34,6 +35,10 @@ public class WebSocketServerHandler extends SimpleChannelInboundHandler<TextWebS
 
     private static final ObjectMapper objectMapper = new ObjectMapper();
 
+    // 用户 -> 上次提示客服分配的时间
+    private static final ConcurrentHashMap<String, Long> userLastAllocationNotice = new ConcurrentHashMap<>();
+    // 允许重复提示的间隔（毫秒）
+    private static final long ALLOCATION_NOTICE_INTERVAL = 5 * 60 * 1000; // 5分钟
     private static final int USER_TIMEOUT = 180;   // 秒
     private static final int ADMIN_TIMEOUT = 300; // 秒
 
@@ -73,47 +78,111 @@ public class WebSocketServerHandler extends SimpleChannelInboundHandler<TextWebS
 
     @Override
     public void handlerRemoved(ChannelHandlerContext ctx) {
-        userChannels.entrySet().removeIf(entry -> entry.getValue().equals(ctx.channel()));
-        adminChannels.entrySet().removeIf(entry -> entry.getValue().equals(ctx.channel()));
+        handleChannelDisconnect(ctx);
         System.out.println("连接断开：" + ctx.channel().id().asShortText());
     }
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-        userChannels.entrySet().removeIf(entry -> entry.getValue().equals(ctx.channel()));
-        adminChannels.entrySet().removeIf(entry -> entry.getValue().equals(ctx.channel()));
+        handleChannelDisconnect(ctx);
         super.channelInactive(ctx);
     }
 
+    /**
+     * 处理通道断开连接 - 统一处理用户和客服的离线逻辑
+     */
+    private void handleChannelDisconnect(ChannelHandlerContext ctx) {
+        // 先检查是否是用户连接断开
+        String userId = getUserIdByChannel(ctx.channel(), userChannels);
+        if (userId != null) {
+            userChannels.remove(userId);
+            chatService.onUserOffline(userId);
+            // 通知所有客服该用户离线
+            notifyUserOfflineToAdmins(userId);
+            System.out.println("用户 " + userId + " 连接断开，已标记为离线");
+            return;
+        }
+
+        // 再检查是否是客服连接断开
+        String adminId = getUserIdByChannel(ctx.channel(), adminChannels);
+        if (adminId != null) {
+            adminChannels.remove(adminId);
+            chatService.onAdminOffline(adminId);
+            // 通知所有客服该客服离线
+            notifyAdminOfflineToOtherAdmins(adminId);
+            System.out.println("客服 " + adminId + " 连接断开，已标记为离线");
+        }
+    }
+
+    /**
+     * 根据通道获取用户ID
+     */
+    private String getUserIdByChannel(Channel channel, ConcurrentHashMap<String, Channel> channels) {
+        return channels.entrySet().stream()
+                .filter(entry -> entry.getValue().equals(channel))
+                .map(Map.Entry::getKey)
+                .findFirst()
+                .orElse(null);
+    }
+
+    /**
+     * 通知所有客服用户离线
+     */
+    private void notifyUserOfflineToAdmins(String userId) {
+        try {
+            ChatMessageDto offlineMsg = new ChatMessageDto();
+            offlineMsg.setFrom(userId);
+            offlineMsg.setTo(null);
+            offlineMsg.setType("clientOffline");
+            offlineMsg.setContent(userId);
+            offlineMsg.setRole("system");
+            offlineMsg.setTimestamp(System.currentTimeMillis());
+
+            String offlineJson = objectMapper.writeValueAsString(offlineMsg);
+            for (Channel adminChannel : adminChannels.values()) {
+                if (adminChannel != null && adminChannel.isActive()) {
+                    adminChannel.writeAndFlush(new TextWebSocketFrame(offlineJson));
+                }
+            }
+        } catch (Exception e) {
+            System.err.println("通知用户离线失败: " + e.getMessage());
+        }
+    }
+
+    /**
+     * 通知其他客服该客服离线
+     */
+    private void notifyAdminOfflineToOtherAdmins(String adminId) {
+        try {
+            ChatMessageDto offlineMsg = new ChatMessageDto();
+            offlineMsg.setFrom(adminId);
+            offlineMsg.setTo(null);
+            offlineMsg.setType("clientOffline");
+            offlineMsg.setContent(adminId);
+            offlineMsg.setRole("system");
+            offlineMsg.setTimestamp(System.currentTimeMillis());
+
+            String offlineJson = objectMapper.writeValueAsString(offlineMsg);
+            for (Channel adminChannel : adminChannels.values()) {
+                if (adminChannel != null && adminChannel.isActive()) {
+                    adminChannel.writeAndFlush(new TextWebSocketFrame(offlineJson));
+                }
+            }
+        } catch (Exception e) {
+            System.err.println("通知客服离线失败: " + e.getMessage());
+        }
+    }
+
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
         String payload = msg.text();
-        System.out.println("收到消息：" + payload);
 
         ChatMessageDto chatMsg = objectMapper.readValue(payload, ChatMessageDto.class);
+        System.out.println("收到消息: " + chatMsg);
 
         switch (chatMsg.getType()) {
             case "login":
-                if ("admin".equals(chatMsg.getRole())) {
-                    adminChannels.put(chatMsg.getFrom(), ctx.channel());
-                    chatService.onAdminLogin(chatMsg.getFrom());
-                    // 避免重复添加 IdleStateHandler
-                    if (ctx.pipeline().get("idleStateHandler") != null) {
-                        ctx.pipeline().remove("idleStateHandler");
-                    }
-                    ctx.pipeline().addBefore("handler", "idleStateHandler",
-                            new IdleStateHandler(ADMIN_TIMEOUT, 0, 0));
-                    System.out.println("客服上线：" + chatMsg.getFrom());
-                } else {
-                    userChannels.put(chatMsg.getFrom(), ctx.channel());
-                    chatService.onUserLogin(chatMsg.getFrom());
-                    if (ctx.pipeline().get("idleStateHandler") != null) {
-                        ctx.pipeline().remove("idleStateHandler");
-                    }
-                    ctx.pipeline().addBefore("handler", "idleStateHandler",
-                            new IdleStateHandler(USER_TIMEOUT, 0, 0));
-                    System.out.println("用户上线：" + chatMsg.getFrom());
-                }
+                handleLogin(ctx, chatMsg);
                 break;
 
             case "chat":
@@ -122,59 +191,112 @@ public class WebSocketServerHandler extends SimpleChannelInboundHandler<TextWebS
 
             case "heartbeat":
                 // 心跳包不需要额外操作，IdleStateHandler 会自动刷新
+                handleHeartBeat(chatMsg);
                 break;
             case "logout":
-                if ("admin".equals(chatMsg.getRole())) {
-                    adminChannels.remove(chatMsg.getFrom());
-                    chatService.onAdminOffline(chatMsg.getFrom());
-                } else {
-                    userChannels.remove(chatMsg.getFrom());
-                    chatService.onUserOffline(chatMsg.getFrom());
-                }
-                System.out.println(chatMsg.getRole() + " " + chatMsg.getFrom() + " 已下线");
+                handleLogout(chatMsg);
+                break;
+            case "read":
+                handleRead(ctx,chatMsg);
                 break;
+            case "typing":
+            case "typingStop":
+                handleTyping(chatMsg);
+                break;
+//            case "status":
+//                // 这个通常是服务端主动推送，不建议客户端直接发
+//                // 如果你希望客户端也能触发，可以在这里做转发
+//                handleStatus(chatMsg);
+//                break;
 
             default:
                 System.out.println("未知消息类型: " + chatMsg.getType());
         }
     }
 
+    private void handleRead(ChannelHandlerContext ctx, ChatMessageDto chatMsg) {
+        chatService.onReadMessage(chatMsg);
+    }
+
+//    private void handleRead(ChatMessageDto chatMsg) {
+//        chatService.onReadMessage(chatMsg);
+//    }
+
+    private void handleHeartBeat(ChatMessageDto chatMsg) {
+        chatService.onHeartbeat(chatMsg);
+    }
+
+    private void handleLogout(ChatMessageDto chatMsg) throws JsonProcessingException {
+        if ("admin".equals(chatMsg.getRole())) {
+            adminChannels.remove(chatMsg.getFrom());
+            chatService.onAdminOffline(chatMsg.getFrom());
+            for (Channel userChannel : userChannels.values()) {
+                userChannel.writeAndFlush(new TextWebSocketFrame(objectMapper.writeValueAsString(
+                        new ChatMessageDto(chatMsg.getFrom(), null, "clientOffline",
+                                chatMsg.getFrom(), "system", System.currentTimeMillis(), null, null,null))));
+            }
+        } else {
+            userChannels.remove(chatMsg.getFrom());
+            chatService.onUserOffline(chatMsg.getFrom());
+            for (Channel adminChannel : adminChannels.values()) {
+                adminChannel.writeAndFlush(new TextWebSocketFrame(objectMapper.writeValueAsString(
+                        new ChatMessageDto(chatMsg.getFrom(), null, "clientOffline",
+                                chatMsg.getFrom(), "system", System.currentTimeMillis(), null, null,null))));
+            }
+        }
+        System.out.println(chatMsg.getRole() + " " + chatMsg.getFrom() + " 已下线");
+    }
+
+    private void handleLogin(ChannelHandlerContext ctx, ChatMessageDto chatMsg) throws JsonProcessingException {
+        if ("admin".equals(chatMsg.getRole())) {
+            if (!chatService.isClientService(chatMsg.getFrom())) {
+                ctx.close();
+                return;
+            }
+            adminChannels.put(chatMsg.getFrom(), ctx.channel());
+            chatService.onAdminLogin(chatMsg.getFrom());
+            // 避免重复添加 IdleStateHandler
+            if (ctx.pipeline().get("idleStateHandler") != null) {
+                ctx.pipeline().remove("idleStateHandler");
+            }
+            ctx.pipeline().addBefore("handler", "idleStateHandler",
+                    new IdleStateHandler(ADMIN_TIMEOUT, 0, 0));
+            System.out.println("客服上线：" + chatMsg.getFrom());
+        } else {
+            userChannels.put(chatMsg.getFrom(), ctx.channel());
+            chatService.onUserLogin(chatMsg.getFrom());
+
+            for (Channel adminChannel : adminChannels.values()) {
+                adminChannel.writeAndFlush(new TextWebSocketFrame(objectMapper.writeValueAsString(
+                        new ChatMessageDto(chatMsg.getFrom(), null, "clientOnline",
+                                chatMsg.getFrom(), "system", System.currentTimeMillis(), null, null,null))));
+            }
+            if (ctx.pipeline().get("idleStateHandler") != null) {
+                ctx.pipeline().remove("idleStateHandler");
+            }
+            ctx.pipeline().addBefore("handler", "idleStateHandler",
+                    new IdleStateHandler(USER_TIMEOUT, 0, 0));
+            System.out.println("用户上线：" + chatMsg.getFrom());
+        }
+    }
+
     @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
         if (evt instanceof IdleStateEvent) {
-            boolean isUser = userChannels.containsValue(ctx.channel());
-            boolean isAdmin = adminChannels.containsValue(ctx.channel());
-            if (isUser) {
-                String userId = userChannels.entrySet().stream()
-                        .filter(e -> e.getValue().equals(ctx.channel()))
-                        .map(Map.Entry::getKey)
-                        .findFirst()
-                        .orElse(null);
-                chatService.onUserOffline(userId);
-                System.out.println("用户超时，下线：" + ctx.channel().id().asShortText());
-                userChannels.entrySet().removeIf(e -> e.getValue().equals(ctx.channel()));
-                ctx.close();
-            } else if (isAdmin) {
-                String adminId = adminChannels.entrySet().stream()
-                        .filter(e -> e.getValue().equals(ctx.channel()))
-                        .map(Map.Entry::getKey)
-                        .findFirst()
-                        .orElse(null);
-                if (adminId != null) {
-                    chatService.onAdminOffline(adminId);
-                }
-                System.out.println("客服超时，下线：" + ctx.channel().id().asShortText());
-                adminChannels.entrySet().removeIf(e -> e.getValue().equals(ctx.channel()));
-                ctx.close();
-            }
+            System.out.println("连接超时，开始处理离线逻辑：" + ctx.channel().id().asShortText());
+            // 使用统一的断开处理逻辑
+            handleChannelDisconnect(ctx);
+            // 关闭连接
+            ctx.close();
         } else {
             super.userEventTriggered(ctx, evt);
         }
     }
 
 
-
     private void handleChat(ChatMessageDto chatMsg) throws Exception {
+        chatMsg.setTimestamp(System.currentTimeMillis());
+        chatService.onHeartbeat(chatMsg);
         if ("user".equals(chatMsg.getRole())) {
             // 先确定目标客服（如果没有则分配一个）
             String targetAdminId = chatMsg.getTo();
@@ -186,9 +308,31 @@ public class WebSocketServerHandler extends SimpleChannelInboundHandler<TextWebS
             }
 
             if (targetAdminId != null) {
-                // 先落库（会话 + 消息）—— onUserMessage 里会根据 chatMsg.from/chatMsg.to 建会话并保存消息
-                chatService.onUserMessage(chatMsg);
+                // 通知用户xxxx客服为您服务
+                // 只在第一次分配客服，或者超过一定时间没提示时，才发提示
+                long now = System.currentTimeMillis();
+                Long lastNoticeTime = userLastAllocationNotice.get(chatMsg.getFrom());
 
+                boolean shouldNotice = (lastNoticeTime == null) || (now - lastNoticeTime > ALLOCATION_NOTICE_INTERVAL);
+
+                if (shouldNotice) {
+                    Channel userChannel = userChannels.get(chatMsg.getFrom());
+                    if (userChannel != null && userChannel.isActive()) {
+                        userChannel.writeAndFlush(new TextWebSocketFrame(
+                                objectMapper.writeValueAsString(
+                                        new ChatMessageDto(targetAdminId, chatMsg.getFrom(), "adminAllocation",
+                                                "客服" + targetAdminId + "号为您服务", "system",
+                                                now, null, null, null))
+                        ));
+                        userLastAllocationNotice.put(chatMsg.getFrom(), now); // 更新最后提示时间
+                    }
+                }
+                // 先落库（会话 + 消息）—— onUserMessage 里会根据 chatMsg.from/chatMsg.to 建会话并保存消息
+                Map<String, Long> result = chatService.onUserMessage(chatMsg);
+                Long sessionId = result.get("sessionId");
+                Long messageId = result.get("messageId");
+                chatMsg.setSessionId(sessionId);
+                chatMsg.setMessageId(messageId);
                 // 再尝试推送给客服（异步，并带回调处理失败）
                 Channel targetAdmin = adminChannels.get(targetAdminId);
                 if (targetAdmin != null && targetAdmin.isActive()) {
@@ -218,14 +362,17 @@ public class WebSocketServerHandler extends SimpleChannelInboundHandler<TextWebS
                     systemMsg.setFrom("robot");
                     systemMsg.setTo(chatMsg.getFrom());
                     systemMsg.setContent("当前没有客服在线，您可以留下您的联系方式，我们会尽快联系您。");
-                    systemMsg.setTimestamp(System.currentTimeMillis());
                     userChannel.writeAndFlush(new TextWebSocketFrame(objectMapper.writeValueAsString(systemMsg)));
                 }
             }
 
         } else if ("admin".equals(chatMsg.getRole())) {
             // 客服发消息：先落库，再推送给用户（已有失败回调）
-            chatService.onAdminMessage(chatMsg);
+            Map<String, Long> result = chatService.onAdminMessage(chatMsg);
+            Long sessionId = result.get("sessionId");
+            Long messageId = result.get("messageId");
+            chatMsg.setSessionId(sessionId);
+            chatMsg.setMessageId(messageId);
             Channel targetUser = userChannels.get(chatMsg.getTo());
             if (targetUser != null && targetUser.isActive()) {
                 targetUser.writeAndFlush(new TextWebSocketFrame(
@@ -244,6 +391,27 @@ public class WebSocketServerHandler extends SimpleChannelInboundHandler<TextWebS
         }
     }
 
+    private void handleTyping(ChatMessageDto chatMsg) throws JsonProcessingException {
+        // 只做转发，不落库
+        if ("user".equals(chatMsg.getRole())) {
+            // 转发给目标客服
+            Channel adminChannel = adminChannels.get(chatMsg.getTo());
+            if (adminChannel != null && adminChannel.isActive()) {
+                adminChannel.writeAndFlush(new TextWebSocketFrame(
+                        objectMapper.writeValueAsString(chatMsg)
+                ));
+            }
+        } else if ("admin".equals(chatMsg.getRole())) {
+            // 转发给目标用户
+            Channel userChannel = userChannels.get(chatMsg.getTo());
+            if (userChannel != null && userChannel.isActive()) {
+                userChannel.writeAndFlush(new TextWebSocketFrame(
+                        objectMapper.writeValueAsString(chatMsg)
+                ));
+            }
+        }
+    }
+
     /**
      * 从客服列表中选择一个客服（轮询）
      */
@@ -264,4 +432,4 @@ public class WebSocketServerHandler extends SimpleChannelInboundHandler<TextWebS
         ctx.close();
     }
 
-}
+}

zksy-ws/src/main/java/com/zksy/server/WebSocketServerInitializer.java

@@ -9,15 +9,14 @@ import io.netty.handler.codec.http.cors.CorsConfig;
 import io.netty.handler.codec.http.cors.CorsConfigBuilder;
 import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
 import io.netty.handler.stream.ChunkedWriteHandler;
+import lombok.RequiredArgsConstructor;
+import org.springframework.data.redis.core.RedisTemplate;
 
+@RequiredArgsConstructor
 public class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> {
 
     private final ChatService chatService;
-
-    public WebSocketServerInitializer(ChatService chatService) {
-        this.chatService = chatService;
-    }
-
+    private final RedisTemplate<Object,Object> redisTemplate;
 
     @Override
     protected void initChannel(SocketChannel ch) {
@@ -25,8 +24,15 @@ public class WebSocketServerInitializer extends ChannelInitializer<SocketChannel
                 .addLast(new HttpServerCodec())
                 .addLast(new HttpObjectAggregator(65536))
                 .addLast(new ChunkedWriteHandler())
-                .addLast(new AuthHandshakeHandler())
-                .addLast(new WebSocketServerProtocolHandler("/ws"))
+                .addLast(new AuthHandshakeHandler(redisTemplate))
+                .addLast(new WebSocketServerProtocolHandler(
+                        "/ws",     // 基础路径
+                        null,      // subprotocols
+                        true,      // allowExtensions
+                        65536,     // maxFrameSize
+                        false,     // allowMaskMismatch
+                        true       // checkStartsWith —— 开启前缀匹配
+                ))
                 // 给业务 handler 起个名字，方便动态 addBefore
                 .addLast("handler", new WebSocketServerHandler(chatService));
     }