diff --git a/lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/NettyClientApplication.java b/lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/NettyClientApplication.java index 13d4254c..2990fbea 100644 --- a/lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/NettyClientApplication.java +++ b/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); + } + } diff --git a/lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/client/NettyClient.java b/lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/client/NettyClient.java new file mode 100644 index 00000000..53470abf --- /dev/null +++ b/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(); + } + +} diff --git a/lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/NettyServerHandlerInitializer.java b/lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/client/NettyClientHandlerInitializer.java similarity index 85% rename from lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/NettyServerHandlerInitializer.java rename to lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/java/cn/iocoder/springboot/lab67/nettyclientdemo/client/NettyClientHandlerInitializer.java index b7a374e4..0215404e 100644 --- a/lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/NettyServerHandlerInitializer.java +++ b/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 { +public class NettyClientHandlerInitializer extends ChannelInitializer { @Autowired private MessageDispatcher messageDispatcher; @@ -23,7 +23,7 @@ public class NettyServerHandlerInitializer extends ChannelInitializer { .addLast(new InvocationEncoder()) // 解码器 .addLast(new InvocationDecoder()) - // 服务端处理器 + // 客户端处理器 .addLast(messageDispatcher) ; } diff --git a/lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/resources/application.yml b/lab-67/lab-67-netty-demo/lab-67-netty-demo-client/src/main/resources/application.yml new file mode 100644 index 00000000..472ca627 --- /dev/null +++ b/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 端口 diff --git a/lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/NettyChannelManager.java b/lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/NettyChannelManager.java new file mode 100644 index 00000000..ae02a600 --- /dev/null +++ b/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 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()); + } + +} diff --git a/lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/NettyServer.java b/lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/NettyServer.java index b1c2449f..d7a3ca9f 100644 --- a/lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/NettyServer.java +++ b/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; diff --git a/lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/handler/NettyServerHandler.java b/lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/handler/NettyServerHandler.java new file mode 100644 index 00000000..de98d43c --- /dev/null +++ b/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(); + } + +} diff --git a/lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/handler/NettyServerHandlerInitializer.java b/lab-67/lab-67-netty-demo/lab-67-netty-demo-server/src/main/java/cn/iocoder/springboot/lab67/nettyserverdemo/server/handler/NettyServerHandlerInitializer.java new file mode 100644 index 00000000..2914ceef --- /dev/null +++ b/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 { + + /** + * 心跳超时时间 + */ + 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) + ; + } + +}