Netty 入门示例 - 完成 Server 连接管理功能

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

@@ -1,4 +1,13 @@
 package cn.iocoder.springboot.lab67.nettyclientdemo;
 
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
 public class NettyClientApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(NettyClientApplication.class, args);
+    }
+
 }

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

@@ -0,0 +1,75 @@
+package cn.iocoder.springboot.lab67.nettyclientdemo.client;
+
+import io.netty.bootstrap.Bootstrap;
+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.NioSocketChannel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+
+@Component
+public class NettyClient {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Value("${netty.server.host}")
+    private String serverHost;
+    @Value("${netty.server.port}")
+    private Integer serverPort;
+
+    @Autowired
+    private NettyClientHandlerInitializer nettyClientHandlerInitializer;
+
+    /**
+     * 线程组，用于客户端对服务端的链接、数据读写
+     */
+    private EventLoopGroup eventGroup = new NioEventLoopGroup();
+    /**
+     * Netty Client Channel
+     */
+    private Channel channel;
+
+    /**
+     * 启动 Netty Server
+     */
+    @PostConstruct
+    public void start() throws InterruptedException {
+        // 创建 Bootstrap 对象，用于 Netty Client 启动
+        Bootstrap bootstrap = new Bootstrap();
+        bootstrap.group(eventGroup) // 设置一个 EventLoopGroup 对象
+                .channel(NioSocketChannel.class)  // 指定 Channel 为客户端 NioSocketChannel
+                .remoteAddress(serverHost, serverPort) // 指定链接服务器的地址
+                .option(ChannelOption.SO_KEEPALIVE, true) // TCP Keepalive 机制，实现 TCP 层级的心跳保活功能
+                .option(ChannelOption.TCP_NODELAY, true) // 允许较小的数据包的发送，降低延迟
+                .handler(nettyClientHandlerInitializer);
+        // 链接服务器，并同步等待成功，即启动客户端
+        ChannelFuture future = bootstrap.connect().sync();
+        if (future.isSuccess()) {
+            channel = future.channel();
+            logger.info("[start][Netty Client 链接服务器({}:{}) 成功]", serverHost, serverPort);
+        }
+    }
+
+    /**
+     * 关闭 Netty Server
+     */
+    @PreDestroy
+    public void shutdown() {
+        // 关闭 Netty Client
+        if (channel != null) {
+            channel.close();
+        }
+        // 优雅关闭一个 EventLoopGroup 对象
+        eventGroup.shutdownGracefully();
+    }
+
+}

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

@@ -1,4 +1,4 @@
-package cn.iocoder.springboot.lab67.nettyserverdemo.server;
+package cn.iocoder.springboot.lab67.nettyclientdemo.client;
 
 import cn.iocoder.springboot.lab67.nettycommondemo.codec.InvocationDecoder;
 import cn.iocoder.springboot.lab67.nettycommondemo.codec.InvocationEncoder;
@@ -9,7 +9,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 @Component
-public class NettyServerHandlerInitializer extends ChannelInitializer<Channel> {
+public class NettyClientHandlerInitializer extends ChannelInitializer<Channel> {
 
     @Autowired
     private MessageDispatcher messageDispatcher;
@@ -23,7 +23,7 @@ public class NettyServerHandlerInitializer extends ChannelInitializer<Channel> {
                 .addLast(new InvocationEncoder())
                 // 解码器
                 .addLast(new InvocationDecoder())
-                // 服务端处理器
+                // 客户端处理器
                 .addLast(messageDispatcher)
         ;
     }

lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/resources/application.yml

@@ -0,0 +1,4 @@
+netty:
+  server:
+    host: 127.0.0.1 # Netty Server 地址
+    port: 8888 # Netty Server 端口

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

@@ -0,0 +1,29 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.server;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelId;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+@Component
+public class NettyChannelManager {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    private ConcurrentMap<ChannelId, Channel> channels = new ConcurrentHashMap<>();
+
+    public void add(Channel channel) {
+        channels.put(channel.id(), channel);
+        logger.info("[add][一个新的连接({})加入]", channel.id());
+    }
+
+    public void remove(Channel channel) {
+        channels.remove(channel.id());
+        logger.info("[add][一个连接({})离开]", channel.id());
+    }
+
+}

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

@@ -1,10 +1,8 @@
 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.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.EventLoopGroup;
+import io.netty.channel.*;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
 import org.slf4j.Logger;

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

@@ -0,0 +1,40 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.server.handler;
+
+import cn.iocoder.springboot.lab67.nettyserverdemo.server.NettyChannelManager;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+public class NettyServerHandler extends ChannelInboundHandlerAdapter {
+
+    private Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Autowired
+    private NettyChannelManager channelManager;
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) {
+        // 从管理器中添加
+        channelManager.add(ctx.channel());
+    }
+
+    @Override
+    public void channelUnregistered(ChannelHandlerContext ctx) {
+        // 从管理器中移除
+        channelManager.remove(ctx.channel());
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        logger.error("[exceptionCaught][连接({}) 发生异常]", ctx.channel().id());
+        // 从管理器中移除
+        channelManager.remove(ctx.channel());
+        // 断开连接
+        ctx.channel().close();
+    }
+
+}

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

@@ -0,0 +1,43 @@
+package cn.iocoder.springboot.lab67.nettyserverdemo.server.handler;
+
+import cn.iocoder.springboot.lab67.nettycommondemo.codec.InvocationDecoder;
+import cn.iocoder.springboot.lab67.nettycommondemo.codec.InvocationEncoder;
+import cn.iocoder.springboot.lab67.nettycommondemo.dispacher.MessageDispatcher;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.handler.timeout.ReadTimeoutHandler;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.TimeUnit;
+
+@Component
+public class NettyServerHandlerInitializer extends ChannelInitializer<Channel> {
+
+    /**
+     * 心跳超时时间
+     */
+    private static final Integer READ_TIMEOUT_SECONDS = 1 * 10;
+
+    @Autowired
+    private MessageDispatcher messageDispatcher;
+    @Autowired
+    private NettyServerHandler nettyServerHandler;
+
+    @Override
+    protected void initChannel(Channel ch) throws Exception {
+        ch.pipeline()
+                // 空闲检测
+                .addLast(new ReadTimeoutHandler(READ_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+                // 编码器
+                .addLast(new InvocationEncoder())
+                // 解码器
+                .addLast(new InvocationDecoder())
+                // 消息分发器
+                .addLast(messageDispatcher)
+                // 服务端处理器
+                .addLast(nettyServerHandler)
+        ;
+    }
+
+}