Netty 入门示例 - 增加业务逻辑

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/client/NettyClient.java

@@ -1,6 +1,7 @@
 package cn.iocoder.springboot.lab67.nettyclientdemo.client;
 
 import cn.iocoder.springboot.lab67.nettyclientdemo.client.handler.NettyClientHandlerInitializer;
+import cn.iocoder.springboot.lab67.nettycommondemo.codec.Invocation;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.*;
 import io.netty.channel.nio.NioEventLoopGroup;
@@ -86,6 +87,7 @@ public class NettyClient {
                 }
             }
         }, RECONNECT_SECONDS, TimeUnit.SECONDS);
+        logger.info("[reconnect][{} 秒后将发起重连]", RECONNECT_SECONDS);
     }
 
     /**
@@ -101,4 +103,22 @@ public class NettyClient {
         eventGroup.shutdownGracefully();
     }
 
+    /**
+     * 发送消息
+     *
+     * @param invocation 消息体
+     */
+    public void send(Invocation invocation) {
+        if (channel == null) {
+            logger.error("[send][连接不存在]");
+            return;
+        }
+        if (!channel.isActive()) {
+            logger.error("[send][连接({})未激活]", channel.id());
+            return;
+        }
+        // 发送消息
+        channel.writeAndFlush(invocation);
+    }
+
 }

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/client/handler/NettyClientHandler.java

@@ -2,8 +2,9 @@ package cn.iocoder.springboot.lab67.nettyclientdemo.client.handler;
 
 import cn.iocoder.springboot.lab67.nettyclientdemo.client.NettyClient;
 import cn.iocoder.springboot.lab67.nettycommondemo.codec.Invocation;
-import cn.iocoder.springboot.lab67.nettycommondemo.heartbeat.HeartbeatRequest;
+import cn.iocoder.springboot.lab67.nettyclientdemo.message.heartbeat.HeartbeatRequest;
 import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.timeout.IdleStateEvent;
@@ -13,6 +14,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
+@ChannelHandler.Sharable
 public class NettyClientHandler extends ChannelInboundHandlerAdapter {
 
     private Logger logger = LoggerFactory.getLogger(getClass());

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/client/handler/NettyClientHandlerInitializer.java

@@ -15,7 +15,7 @@ public class NettyClientHandlerInitializer extends ChannelInitializer<Channel> {
     /**
      * 心跳超时时间
      */
-    private static final Integer WRITE_TIMEOUT_SECONDS = 3;
+    private static final Integer READ_TIMEOUT_SECONDS = 60;
 
     @Autowired
     private MessageDispatcher messageDispatcher;
@@ -24,15 +24,15 @@ public class NettyClientHandlerInitializer extends ChannelInitializer<Channel> {
     private NettyClientHandler nettyClientHandler;
 
     @Override
-    protected void initChannel(Channel ch) throws Exception {
+    protected void initChannel(Channel ch) {
         ch.pipeline()
                 // 空闲检测
-                .addLast(new IdleStateHandler(WRITE_TIMEOUT_SECONDS, 0, 0))
+                .addLast(new IdleStateHandler(READ_TIMEOUT_SECONDS, 0, 0))
                 // 编码器
                 .addLast(new InvocationEncoder())
                 // 解码器
                 .addLast(new InvocationDecoder())
-                // 客户端处理器
+                // 消息分发器
                 .addLast(messageDispatcher)
                 // 客户端处理器
                 .addLast(nettyClientHandler)

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/controller/TestController.java

@@ -0,0 +1,26 @@
+package cn.iocoder.springboot.lab67.nettyclientdemo.controller;
+
+import cn.iocoder.springboot.lab67.nettyclientdemo.client.NettyClient;
+import cn.iocoder.springboot.lab67.nettycommondemo.codec.Invocation;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/test")
+public class TestController {
+
+    @Autowired
+    private NettyClient nettyClient;
+
+    @PostMapping("/mock")
+    public String mock(String type, String message) {
+        // 创建 Invocation 对象
+        Invocation invocation = new Invocation(type, message);
+        // 发送消息
+        nettyClient.send(invocation);
+        return "success";
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/message/auth/AuthRequest.java

@@ -0,0 +1,33 @@
+package cn.iocoder.springboot.lab67.nettyclientdemo.message.auth;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 用户认证请求
+ */
+public class AuthRequest implements Message {
+
+    public static final String TYPE = "AUTH_REQUEST";
+
+    /**
+     * 认证 Token
+     */
+    private String accessToken;
+
+    public String getAccessToken() {
+        return accessToken;
+    }
+
+    public AuthRequest setAccessToken(String accessToken) {
+        this.accessToken = accessToken;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "AuthRequest{" +
+                "accessToken='" + accessToken + '\'' +
+                '}';
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/message/auth/AuthResponse.java

@@ -0,0 +1,47 @@
+package cn.iocoder.springboot.lab67.nettyclientdemo.message.auth;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 用户认证响应
+ */
+public class AuthResponse implements Message {
+
+    public static final String TYPE = "AUTH_RESPONSE";
+
+    /**
+     * 响应状态码
+     */
+    private Integer code;
+    /**
+     * 响应提示
+     */
+    private String message;
+
+    public Integer getCode() {
+        return code;
+    }
+
+    public AuthResponse setCode(Integer code) {
+        this.code = code;
+        return this;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public AuthResponse setMessage(String message) {
+        this.message = message;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "AuthResponse{" +
+                "code=" + code +
+                ", message='" + message + '\'' +
+                '}';
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/message/chat/ChatRedirectToUserRequest.java

@@ -0,0 +1,47 @@
+package cn.iocoder.springboot.lab67.nettyclientdemo.message.chat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 转发消息给一个用户的 Message
+ */
+public class ChatRedirectToUserRequest implements Message {
+
+    public static final String TYPE = "CHAT_REDIRECT_TO_USER_REQUEST";
+
+    /**
+     * 消息编号
+     */
+    private String msgId;
+    /**
+     * 内容
+     */
+    private String content;
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public ChatRedirectToUserRequest setMsgId(String msgId) {
+        this.msgId = msgId;
+        return this;
+    }
+
+    public String getContent() {
+        return content;
+    }
+
+    public ChatRedirectToUserRequest setContent(String content) {
+        this.content = content;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "ChatRedirectToUserRequest{" +
+                "msgId='" + msgId + '\'' +
+                ", content='" + content + '\'' +
+                '}';
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/message/chat/ChatSendResponse.java

@@ -0,0 +1,60 @@
+package cn.iocoder.springboot.lab67.nettyclientdemo.message.chat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 聊天发送消息结果的 Response
+ */
+public class ChatSendResponse implements Message {
+
+    public static final String TYPE = "CHAT_SEND_RESPONSE";
+
+    /**
+     * 消息编号
+     */
+    private String msgId;
+    /**
+     * 响应状态码
+     */
+    private Integer code;
+    /**
+     * 响应提示
+     */
+    private String message;
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public ChatSendResponse setMsgId(String msgId) {
+        this.msgId = msgId;
+        return this;
+    }
+
+    public Integer getCode() {
+        return code;
+    }
+
+    public ChatSendResponse setCode(Integer code) {
+        this.code = code;
+        return this;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public ChatSendResponse setMessage(String message) {
+        this.message = message;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "ChatSendResponse{" +
+                "msgId='" + msgId + '\'' +
+                ", code=" + code +
+                ", message='" + message + '\'' +
+                '}';
+    }
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/message/chat/ChatSendToAllRequest.java

@@ -0,0 +1,47 @@
+package cn.iocoder.springboot.lab67.nettyclientdemo.message.chat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 发送给所有人的群聊消息的 Message
+ */
+public class ChatSendToAllRequest implements Message {
+
+    public static final String TYPE = "CHAT_SEND_TO_ALL_REQUEST";
+
+    /**
+     * 消息编号
+     */
+    private String msgId;
+    /**
+     * 内容
+     */
+    private String content;
+
+    public String getContent() {
+        return content;
+    }
+
+    public ChatSendToAllRequest setContent(String content) {
+        this.content = content;
+        return this;
+    }
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public ChatSendToAllRequest setMsgId(String msgId) {
+        this.msgId = msgId;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "ChatSendToAllRequest{" +
+                "msgId='" + msgId + '\'' +
+                ", content='" + content + '\'' +
+                '}';
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/message/chat/ChatSendToOneRequest.java

@@ -0,0 +1,61 @@
+package cn.iocoder.springboot.lab67.nettyclientdemo.message.chat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 发送给指定人的私聊消息 Request
+ */
+public class ChatSendToOneRequest implements Message {
+
+    public static final String TYPE = "CHAT_SEND_TO_ONE_REQUEST";
+
+    /**
+     * 发送给的用户
+     */
+    private String toUser;
+    /**
+     * 消息编号
+     */
+    private String msgId;
+    /**
+     * 内容
+     */
+    private String content;
+
+    public String getToUser() {
+        return toUser;
+    }
+
+    public ChatSendToOneRequest setToUser(String toUser) {
+        this.toUser = toUser;
+        return this;
+    }
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public ChatSendToOneRequest setMsgId(String msgId) {
+        this.msgId = msgId;
+        return this;
+    }
+
+    public String getContent() {
+        return content;
+    }
+
+    public ChatSendToOneRequest setContent(String content) {
+        this.content = content;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "ChatSendToOneRequest{" +
+                "toUser='" + toUser + '\'' +
+                ", msgId='" + msgId + '\'' +
+                ", content='" + content + '\'' +
+                '}';
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/message/heartbeat/HeartbeatRequest.java

@@ -1,4 +1,4 @@
-package cn.iocoder.springboot.lab67.nettycommondemo.heartbeat;
+package cn.iocoder.springboot.lab67.nettyclientdemo.message.heartbeat;
 
 import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
 
@@ -12,4 +12,9 @@ public class HeartbeatRequest implements Message {
      */
     public static final String TYPE = "HEARTBEAT_REQUEST";
 
+    @Override
+    public String toString() {
+        return "HeartbeatRequest{}";
+    }
+
 }

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/message/heartbeat/HeartbeatResponse.java

@@ -1,4 +1,4 @@
-package cn.iocoder.springboot.lab67.nettycommondemo.heartbeat;
+package cn.iocoder.springboot.lab67.nettyclientdemo.message.heartbeat;
 
 import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
 
@@ -12,4 +12,9 @@ public class HeartbeatResponse implements Message {
      */
     public static final String TYPE = "HEARTBEAT_RESPONSE";
 
+    @Override
+    public String toString() {
+        return "HeartbeatResponse{}";
+    }
+
 }

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/messagehandler/auth/AuthResponseHandler.java

@@ -0,0 +1,25 @@
+package cn.iocoder.springboot.lab67.nettyclientdemo.messagehandler.auth;
+
+import cn.iocoder.springboot.lab67.nettyclientdemo.message.auth.AuthResponse;
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.MessageHandler;
+import io.netty.channel.Channel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+@Component
+public class AuthResponseHandler implements MessageHandler<AuthResponse> {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public void execute(Channel channel, AuthResponse message) {
+        logger.info("[execute][认证结果：{}]", message);
+    }
+
+    @Override
+    public String getType() {
+        return AuthResponse.TYPE;
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/messagehandler/chat/ChatRedirectToUserRequestHandler.java

@@ -0,0 +1,25 @@
+package cn.iocoder.springboot.lab67.nettyclientdemo.messagehandler.chat;
+
+import cn.iocoder.springboot.lab67.nettyclientdemo.message.chat.ChatRedirectToUserRequest;
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.MessageHandler;
+import io.netty.channel.Channel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ChatRedirectToUserRequestHandler implements MessageHandler<ChatRedirectToUserRequest> {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public void execute(Channel channel, ChatRedirectToUserRequest message) {
+        logger.info("[execute][收到消息：{}]", message);
+    }
+
+    @Override
+    public String getType() {
+        return ChatRedirectToUserRequest.TYPE;
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/messagehandler/chat/ChatSendResponseHandler.java

@@ -0,0 +1,25 @@
+package cn.iocoder.springboot.lab67.nettyclientdemo.messagehandler.chat;
+
+import cn.iocoder.springboot.lab67.nettyclientdemo.message.chat.ChatSendResponse;
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.MessageHandler;
+import io.netty.channel.Channel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ChatSendResponseHandler implements MessageHandler<ChatSendResponse> {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public void execute(Channel channel, ChatSendResponse message) {
+        logger.info("[execute][发送结果：{}]", message);
+    }
+
+    @Override
+    public String getType() {
+        return ChatSendResponse.TYPE;
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/messagehandler/heartbeat/HeartbeatResponseHandler.java

@@ -1,7 +1,7 @@
-package cn.iocoder.springboot.lab67.nettyclientdemo.messagehandler;
+package cn.iocoder.springboot.lab67.nettyclientdemo.messagehandler.heartbeat;
 
 import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.MessageHandler;
-import cn.iocoder.springboot.lab67.nettycommondemo.heartbeat.HeartbeatResponse;
+import cn.iocoder.springboot.lab67.nettyclientdemo.message.heartbeat.HeartbeatResponse;
 import io.netty.channel.Channel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

lab-67/lab-67-netty-demo/lab-67-netty-demo-common/src/main/java/cn/iocoder/springboot/lab67/nettycommondemo/codec/Invocation.java

@@ -17,6 +17,11 @@ public class Invocation {
     public Invocation() {
     }
 
+    public Invocation(String type, String message) {
+        this.type = type;
+        this.message = message;
+    }
+
     public Invocation(String type, Message message) {
         this.type = type;
         this.message = JSON.toJSONString(message);
@@ -40,4 +45,11 @@ public class Invocation {
         return this;
     }
 
+    @Override
+    public String toString() {
+        return "Invocation{" +
+                "type='" + type + '\'' +
+                ", message='" + message + '\'' +
+                '}';
+    }
 }

lab-67/lab-67-netty-demo/lab-67-netty-demo-common/src/main/java/cn/iocoder/springboot/lab67/nettycommondemo/codec/InvocationDecoder.java

@@ -5,11 +5,15 @@ import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.handler.codec.CorruptedFrameException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.List;
 
 public class InvocationDecoder extends ByteToMessageDecoder {
 
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
     @Override
     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
         // 标记当前读取位置
@@ -34,6 +38,7 @@ public class InvocationDecoder extends ByteToMessageDecoder {
         // 解析成 Invocation
         Invocation invocation = JSON.parseObject(content, Invocation.class);
         out.add(invocation);
+        logger.info("[decode][连接({}) 解析到一条消息({})]", ctx.channel().id(), invocation.toString());
     }
 
 }

lab-67/lab-67-netty-demo/lab-67-netty-demo-common/src/main/java/cn/iocoder/springboot/lab67/nettycommondemo/codec/InvocationEncoder.java

@@ -4,9 +4,13 @@ import com.alibaba.fastjson.JSON;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.MessageToByteEncoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class InvocationEncoder extends MessageToByteEncoder<Invocation> {
 
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
     @Override
     protected void encode(ChannelHandlerContext ctx, Invocation invocation, ByteBuf out) {
         // 将 Invocation 转换成 byte[] 数组
@@ -15,6 +19,7 @@ public class InvocationEncoder extends MessageToByteEncoder<Invocation> {
         out.writeInt(content.length);
         // 写入内容
         out.writeBytes(content);
+        logger.info("[decode][连接({}) 编码了一条消息({})]", ctx.channel().id(), invocation.toString());
     }
 
 }

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/message/auth/AuthRequest.java

@@ -0,0 +1,33 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.message.auth;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 用户认证请求
+ */
+public class AuthRequest implements Message {
+
+    public static final String TYPE = "AUTH_REQUEST";
+
+    /**
+     * 认证 Token
+     */
+    private String accessToken;
+
+    public String getAccessToken() {
+        return accessToken;
+    }
+
+    public AuthRequest setAccessToken(String accessToken) {
+        this.accessToken = accessToken;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "AuthRequest{" +
+                "accessToken='" + accessToken + '\'' +
+                '}';
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/message/auth/AuthResponse.java

@@ -0,0 +1,47 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.message.auth;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 用户认证响应
+ */
+public class AuthResponse implements Message {
+
+    public static final String TYPE = "AUTH_RESPONSE";
+
+    /**
+     * 响应状态码
+     */
+    private Integer code;
+    /**
+     * 响应提示
+     */
+    private String message;
+
+    public Integer getCode() {
+        return code;
+    }
+
+    public AuthResponse setCode(Integer code) {
+        this.code = code;
+        return this;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public AuthResponse setMessage(String message) {
+        this.message = message;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "AuthResponse{" +
+                "code=" + code +
+                ", message='" + message + '\'' +
+                '}';
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/message/chat/ChatRedirectToUserRequest.java

@@ -0,0 +1,47 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.message.chat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 转发消息给一个用户的 Message
+ */
+public class ChatRedirectToUserRequest implements Message {
+
+    public static final String TYPE = "CHAT_REDIRECT_TO_USER_REQUEST";
+
+    /**
+     * 消息编号
+     */
+    private String msgId;
+    /**
+     * 内容
+     */
+    private String content;
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public ChatRedirectToUserRequest setMsgId(String msgId) {
+        this.msgId = msgId;
+        return this;
+    }
+
+    public String getContent() {
+        return content;
+    }
+
+    public ChatRedirectToUserRequest setContent(String content) {
+        this.content = content;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "ChatRedirectToUserRequest{" +
+                "msgId='" + msgId + '\'' +
+                ", content='" + content + '\'' +
+                '}';
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/message/chat/ChatSendResponse.java

@@ -0,0 +1,60 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.message.chat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 聊天发送消息结果的 Response
+ */
+public class ChatSendResponse implements Message {
+
+    public static final String TYPE = "CHAT_SEND_RESPONSE";
+
+    /**
+     * 消息编号
+     */
+    private String msgId;
+    /**
+     * 响应状态码
+     */
+    private Integer code;
+    /**
+     * 响应提示
+     */
+    private String message;
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public ChatSendResponse setMsgId(String msgId) {
+        this.msgId = msgId;
+        return this;
+    }
+
+    public Integer getCode() {
+        return code;
+    }
+
+    public ChatSendResponse setCode(Integer code) {
+        this.code = code;
+        return this;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public ChatSendResponse setMessage(String message) {
+        this.message = message;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "ChatSendResponse{" +
+                "msgId='" + msgId + '\'' +
+                ", code=" + code +
+                ", message='" + message + '\'' +
+                '}';
+    }
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/message/chat/ChatSendToAllRequest.java

@@ -0,0 +1,47 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.message.chat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 发送给所有人的群聊消息的 Message
+ */
+public class ChatSendToAllRequest implements Message {
+
+    public static final String TYPE = "CHAT_SEND_TO_ALL_REQUEST";
+
+    /**
+     * 消息编号
+     */
+    private String msgId;
+    /**
+     * 内容
+     */
+    private String content;
+
+    public String getContent() {
+        return content;
+    }
+
+    public ChatSendToAllRequest setContent(String content) {
+        this.content = content;
+        return this;
+    }
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public ChatSendToAllRequest setMsgId(String msgId) {
+        this.msgId = msgId;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "ChatSendToAllRequest{" +
+                "msgId='" + msgId + '\'' +
+                ", content='" + content + '\'' +
+                '}';
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/message/chat/ChatSendToOneRequest.java

@@ -0,0 +1,61 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.message.chat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 发送给指定人的私聊消息 Request
+ */
+public class ChatSendToOneRequest implements Message {
+
+    public static final String TYPE = "CHAT_SEND_TO_ONE_REQUEST";
+
+    /**
+     * 发送给的用户
+     */
+    private String toUser;
+    /**
+     * 消息编号
+     */
+    private String msgId;
+    /**
+     * 内容
+     */
+    private String content;
+
+    public String getToUser() {
+        return toUser;
+    }
+
+    public ChatSendToOneRequest setToUser(String toUser) {
+        this.toUser = toUser;
+        return this;
+    }
+
+    public String getMsgId() {
+        return msgId;
+    }
+
+    public ChatSendToOneRequest setMsgId(String msgId) {
+        this.msgId = msgId;
+        return this;
+    }
+
+    public String getContent() {
+        return content;
+    }
+
+    public ChatSendToOneRequest setContent(String content) {
+        this.content = content;
+        return this;
+    }
+
+    @Override
+    public String toString() {
+        return "ChatSendToOneRequest{" +
+                "toUser='" + toUser + '\'' +
+                ", msgId='" + msgId + '\'' +
+                ", content='" + content + '\'' +
+                '}';
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/message/heartbeat/HeartbeatRequest.java

@@ -0,0 +1,20 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.message.heartbeat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 消息 - 心跳请求
+ */
+public class HeartbeatRequest implements Message {
+
+    /**
+     * 类型 - 心跳请求
+     */
+    public static final String TYPE = "HEARTBEAT_REQUEST";
+
+    @Override
+    public String toString() {
+        return "HeartbeatRequest{}";
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/message/heartbeat/HeartbeatResponse.java

@@ -0,0 +1,20 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.message.heartbeat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.Message;
+
+/**
+ * 消息 - 心跳响应
+ */
+public class HeartbeatResponse implements Message {
+
+    /**
+     * 类型 - 心跳响应
+     */
+    public static final String TYPE = "HEARTBEAT_RESPONSE";
+
+    @Override
+    public String toString() {
+        return "HeartbeatResponse{}";
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/messagehandler/auth/AuthRequestHandler.java

@@ -0,0 +1,42 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.messagehandler.auth;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.codec.Invocation;
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.MessageHandler;
+import cn.iocoder.springboot.lab67.nettyserverdemo.message.auth.AuthRequest;
+import cn.iocoder.springboot.lab67.nettyserverdemo.message.auth.AuthResponse;
+import cn.iocoder.springboot.lab67.nettyserverdemo.server.NettyChannelManager;
+import io.netty.channel.Channel;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+
+@Component
+public class AuthRequestHandler implements MessageHandler<AuthRequest> {
+
+    @Autowired
+    private NettyChannelManager nettyChannelManager;
+
+    @Override
+    public void execute(Channel channel, AuthRequest authRequest) {
+        // 如果未传递 accessToken
+        if (StringUtils.isEmpty(authRequest.getAccessToken())) {
+            AuthResponse authResponse = new AuthResponse().setCode(1).setMessage("认证 accessToken 未传入");
+            channel.writeAndFlush(new Invocation(AuthResponse.TYPE, authResponse));
+            return;
+        }
+
+        // 添加到 WebSocketUtil 中
+        // 考虑到代码简化，我们先直接使用 accessToken 作为 User
+        nettyChannelManager.addUser(channel, authRequest.getAccessToken());
+
+        // 判断是否认证成功。这里，假装直接成功
+        AuthResponse authResponse = new AuthResponse().setCode(0);
+        channel.writeAndFlush(new Invocation(AuthResponse.TYPE, authResponse));
+    }
+
+    @Override
+    public String getType() {
+        return AuthRequest.TYPE;
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/messagehandler/chat/ChatSendToAllHandler.java

@@ -0,0 +1,36 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.messagehandler.chat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.codec.Invocation;
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.MessageHandler;
+import cn.iocoder.springboot.lab67.nettyserverdemo.message.chat.ChatSendResponse;
+import cn.iocoder.springboot.lab67.nettyserverdemo.message.chat.ChatSendToAllRequest;
+import cn.iocoder.springboot.lab67.nettyserverdemo.message.chat.ChatRedirectToUserRequest;
+import cn.iocoder.springboot.lab67.nettyserverdemo.server.NettyChannelManager;
+import io.netty.channel.Channel;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ChatSendToAllHandler implements MessageHandler<ChatSendToAllRequest> {
+
+    @Autowired
+    private NettyChannelManager nettyChannelManager;
+
+    @Override
+    public void execute(Channel channel, ChatSendToAllRequest message) {
+        // 这里，假装直接成功
+        ChatSendResponse sendResponse = new ChatSendResponse().setMsgId(message.getMsgId()).setCode(0);
+        channel.writeAndFlush(new Invocation(ChatSendResponse.TYPE, sendResponse));
+
+        // 创建转发的消息，并广播发送
+        ChatRedirectToUserRequest sendToUserRequest = new ChatRedirectToUserRequest().setMsgId(message.getMsgId())
+                .setContent(message.getContent());
+        nettyChannelManager.sendAll(new Invocation(ChatRedirectToUserRequest.TYPE, sendToUserRequest));
+    }
+
+    @Override
+    public String getType() {
+        return ChatSendToAllRequest.TYPE;
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/messagehandler/chat/ChatSendToOneHandler.java

@@ -0,0 +1,36 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.messagehandler.chat;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.codec.Invocation;
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.MessageHandler;
+import cn.iocoder.springboot.lab67.nettyserverdemo.message.chat.ChatSendResponse;
+import cn.iocoder.springboot.lab67.nettyserverdemo.message.chat.ChatSendToOneRequest;
+import cn.iocoder.springboot.lab67.nettyserverdemo.message.chat.ChatRedirectToUserRequest;
+import cn.iocoder.springboot.lab67.nettyserverdemo.server.NettyChannelManager;
+import io.netty.channel.Channel;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ChatSendToOneHandler implements MessageHandler<ChatSendToOneRequest> {
+
+    @Autowired
+    private NettyChannelManager nettyChannelManager;
+
+    @Override
+    public void execute(Channel channel, ChatSendToOneRequest message) {
+        // 这里，假装直接成功
+        ChatSendResponse sendResponse = new ChatSendResponse().setMsgId(message.getMsgId()).setCode(0);
+        channel.writeAndFlush(new Invocation(ChatSendResponse.TYPE, sendResponse));
+
+        // 创建转发的消息，发送给指定用户
+        ChatRedirectToUserRequest sendToUserRequest = new ChatRedirectToUserRequest().setMsgId(message.getMsgId())
+                .setContent(message.getContent());
+        nettyChannelManager.send(message.getToUser(), new Invocation(ChatRedirectToUserRequest.TYPE, sendToUserRequest));
+    }
+
+    @Override
+    public String getType() {
+        return ChatSendToOneRequest.TYPE;
+    }
+
+}

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/messagehandler/heartbeat/HeartbeatRequestHandler.java

@@ -1,9 +1,9 @@
-package cn.iocoder.springboot.lab67.nettyserverdemo.messagehandler;
+package cn.iocoder.springboot.lab67.nettyserverdemo.messagehandler.heartbeat;
 
 import cn.iocoder.springboot.lab67.nettycommondemo.codec.Invocation;
 import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.MessageHandler;
-import cn.iocoder.springboot.lab67.nettycommondemo.heartbeat.HeartbeatRequest;
-import cn.iocoder.springboot.lab67.nettycommondemo.heartbeat.HeartbeatResponse;
+import cn.iocoder.springboot.lab67.nettyserverdemo.message.heartbeat.HeartbeatRequest;
+import cn.iocoder.springboot.lab67.nettyserverdemo.message.heartbeat.HeartbeatResponse;
 import io.netty.channel.Channel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/NettyChannelManager.java

@@ -1,7 +1,9 @@
 package cn.iocoder.springboot.lab67.nettyserverdemo.server;
 
+import cn.iocoder.springboot.lab67.nettycommondemo.codec.Invocation;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelId;
+import io.netty.util.AttributeKey;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
@@ -12,18 +14,64 @@ import java.util.concurrent.ConcurrentMap;
 @Component
 public class NettyChannelManager {
 
+    private static final AttributeKey<String> CHANNEL_ATTR_KEY_USER = AttributeKey.newInstance("user");
+
     private Logger logger = LoggerFactory.getLogger(getClass());
 
     private ConcurrentMap<ChannelId, Channel> channels = new ConcurrentHashMap<>();
+    private ConcurrentMap<String, Channel> userChannels = new ConcurrentHashMap<>();
 
     public void add(Channel channel) {
         channels.put(channel.id(), channel);
         logger.info("[add][一个连接({})加入]", channel.id());
     }
 
+    public void addUser(Channel channel, String user) {
+        Channel existChannel = channels.get(channel.id());
+        if (existChannel == null) {
+            logger.error("[addUser][连接({}) 不存在]", channel.id());
+            return;
+        }
+        // 设置属性
+        channel.attr(CHANNEL_ATTR_KEY_USER).set(user);
+        // 添加到 userChannels
+        userChannels.put(user, channel);
+    }
+
     public void remove(Channel channel) {
+        // 移除 channels
         channels.remove(channel.id());
+        // 移除 userChannels
+        if (channel.hasAttr(CHANNEL_ATTR_KEY_USER)) {
+            userChannels.remove(channel.attr(CHANNEL_ATTR_KEY_USER).get());
+        }
         logger.info("[remove][一个连接({})离开]", channel.id());
     }
 
+    public void send(String user, Invocation invocation) {
+        // 获得用户对应的 Channel
+        Channel channel = userChannels.get(user);
+        if (channel == null) {
+            logger.error("[send][连接不存在]");
+            return;
+        }
+        if (!channel.isActive()) {
+            logger.error("[send][连接({})未激活]", channel.id());
+            return;
+        }
+        // 发送消息
+        channel.writeAndFlush(invocation);
+    }
+
+    public void sendAll(Invocation invocation) {
+        for (Channel channel : channels.values()) {
+            if (!channel.isActive()) {
+                logger.error("[send][连接({})未激活]", channel.id());
+                return;
+            }
+            // 发送消息
+            channel.writeAndFlush(invocation);
+        }
+    }
+
 }

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/NettyServer.java

@@ -2,7 +2,10 @@ package cn.iocoder.springboot.lab67.nettyserverdemo.server;
 
 import cn.iocoder.springboot.lab67.nettyserverdemo.server.handler.NettyServerHandlerInitializer;
 import io.netty.bootstrap.ServerBootstrap;
-import io.netty.channel.*;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import org.slf4j.Logger;
@@ -25,6 +28,8 @@ public class NettyServer {
 
     @Autowired
     private NettyServerHandlerInitializer nettyServerHandlerInitializer;
+    @Autowired
+    private NettyChannelManager nettyChannelManager;
 
     /**
      * boss 线程组，用于服务端接受客户端的连接

lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/handler/NettyServerHandlerInitializer.java

@@ -17,7 +17,7 @@ public class NettyServerHandlerInitializer extends ChannelInitializer<Channel> {
     /**
      * 心跳超时时间
      */
-    private static final Integer READ_TIMEOUT_SECONDS = 1000;
+    private static final Integer READ_TIMEOUT_SECONDS = 3 * 60;
 
     @Autowired
     private MessageDispatcher messageDispatcher;