以下是一套自带半包处理能力的Netty服务端客户端完整可运行示例完全适配之前我们讨论的TCP无边界流问题实现客户端发送任意长度字符串、服务端都能稳定接收并回显的效果一Netty服务端客户端完整可运行示例第一步先引入Maven依赖确保项目中导入稳定版Netty依赖避免使用处于Alpha阶段的5.0版本dependencygroupIdio.netty/groupIdartifactIdnetty-all/artifactIdversion4.1.100.Final/version/dependency第二步服务端完整代码服务端采用主从Reactor双线程模型提前加入换行拆帧处理器彻底解决半包问题同时配置了标准TCP参数importio.netty.bootstrap.ServerBootstrap;importio.netty.channel.ChannelFuture;importio.netty.channel.ChannelHandlerContext;importio.netty.channel.ChannelInboundHandlerAdapter;importio.netty.channel.ChannelInitializer;importio.netty.channel.nio.NioEventLoopGroup;importio.netty.channel.socket.SocketChannel;importio.netty.channel.socket.nio.NioServerSocketChannel;importio.netty.channel.ChannelOption;importio.netty.handler.codec.LineBasedFrameDecoder;importio.netty.handler.codec.string.StringDecoder;importio.netty.handler.codec.string.StringEncoder;importjava.nio.charset.StandardCharsets;publicclassNettyServer{publicstaticvoidmain(String[]args)throwsInterruptedException{// 1. 初始化双线程组boss组负责接收连接worker组负责处理连接读写NioEventLoopGroupbossGroupnewNioEventLoopGroup(1);NioEventLoopGroupworkerGroupnewNioEventLoopGroup();try{ServerBootstrapserverBootstrapnewServerBootstrap();serverBootstrap.group(bossGroup,workerGroup)// 指定NIO传输模型.channel(NioServerSocketChannel.class)// 服务端连接排队队列最大长度.option(ChannelOption.SO_BACKLOG,1024)// 开启TCP心跳机制检测死连接.childOption(ChannelOption.SO_KEEPALIVE,true)// 关闭Nagle算法保证高实时性场景下数据立即发送.childOption(ChannelOption.TCP_NODELAY,true)// 为每个新连接初始化处理器流水线.childHandler(newChannelInitializerSocketChannel(){OverrideprotectedvoidinitChannel(SocketChannelch){// 核心先加入换行拆帧器最大支持2048字节的单条消息彻底解决半包ch.pipeline().addLast(newLineBasedFrameDecoder(2048));// 加入字符串编解码器自动完成字节和字符串的转换ch.pipeline().addLast(newStringDecoder(StandardCharsets.UTF_8));ch.pipeline().addLast(newStringEncoder(StandardCharsets.UTF_8));// 自定义业务处理器ch.pipeline().addLast(newChannelInboundHandlerAdapter(){OverridepublicvoidchannelRead(ChannelHandlerContextctx,Objectmsg){StringreceiveMsg(String)msg;System.out.println(服务端收到消息receiveMsg);// 给客户端回写响应末尾加换行符匹配拆帧规则ctx.writeAndFlush(服务端已收到receiveMsgSystem.lineSeparator());}OverridepublicvoidexceptionCaught(ChannelHandlerContextctx,Throwablecause){cause.printStackTrace();ctx.close();}});}});// 绑定8888端口同步等待绑定完成ChannelFuturebindFutureserverBootstrap.bind(8888).sync();System.out.println(Netty服务端启动成功监听端口8888);// 阻塞等待服务端通道关闭避免主线程直接退出bindFuture.channel().closeFuture().sync();}finally{// 优雅关闭所有线程组释放资源bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}}第三步客户端完整代码客户端同样配置和服务端匹配的拆帧处理器保证两端的消息边界规则完全对齐避免出现解析异常importio.netty.bootstrap.Bootstrap;importio.netty.channel.ChannelFuture;importio.netty.channel.ChannelHandlerContext;importio.netty.channel.ChannelInboundHandlerAdapter;importio.netty.channel.ChannelInitializer;importio.netty.channel.nio.NioEventLoopGroup;importio.netty.channel.socket.SocketChannel;importio.netty.channel.socket.nio.NioSocketChannel;importio.netty.handler.codec.LineBasedFrameDecoder;importio.netty.handler.codec.string.StringDecoder;importio.netty.handler.codec.string.StringEncoder;importjava.nio.charset.StandardCharsets;publicclassNettyClient{publicstaticvoidmain(String[]args)throwsInterruptedException{// 初始化客户端事件循环组处理所有连接和读写事件NioEventLoopGroupeventLoopGroupnewNioEventLoopGroup();try{BootstrapbootstrapnewBootstrap();bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)// 设置连接超时时间为3秒.option(io.netty.channel.ChannelOption.CONNECT_TIMEOUT_MILLIS,3000).handler(newChannelInitializerSocketChannel(){OverrideprotectedvoidinitChannel(SocketChannelch){// 和服务端配置完全一致的拆帧器保证消息边界对齐ch.pipeline().addLast(newLineBasedFrameDecoder(2048));ch.pipeline().addLast(newStringDecoder(StandardCharsets.UTF_8));ch.pipeline().addLast(newStringEncoder(StandardCharsets.UTF_8));// 自定义客户端业务处理器ch.pipeline().addLast(newChannelInboundHandlerAdapter(){OverridepublicvoidchannelActive(ChannelHandlerContextctx){// 连接建立成功后向服务端发送测试消息末尾加换行符匹配拆帧规则StringtestMsghello,how are youSystem.lineSeparator();ctx.writeAndFlush(testMsg);System.out.println(客户端已发送消息);}OverridepublicvoidchannelRead(ChannelHandlerContextctx,Objectmsg){System.out.println(客户端收到服务端响应msg);}OverridepublicvoidexceptionCaught(ChannelHandlerContextctx,Throwablecause){cause.printStackTrace();ctx.close();}});}});// 异步连接本地8888端口同步等待连接完成ChannelFutureconnectFuturebootstrap.connect(127.0.0.1,8888).sync();// 阻塞等待客户端通道关闭connectFuture.channel().closeFuture().sync();}finally{eventLoopGroup.shutdownGracefully();}}}运行说明先启动服务端控制台打印启动成功提示后再启动客户端两端通过LineBasedFrameDecoder自动处理TCP粘包半包无论消息长短都能稳定解析运行后可以看到服务端完整打印客户端发送的长字符串客户端也能正常收到服务端的回显响应完全复现之前测试短字符串时的正常效果同时支持任意长度的文本消息传输。二Netty服务端与客户端全流程分析Netty基于异步事件驱动模型实现高性能网络通信服务端和客户端的运行流程围绕核心的线程模型、IO事件处理机制展开以下从核心组件、启动流程、运行时通信全链路三个维度进行完整拆解一、核心基础组件前置说明在分析流程前先明确几个贯穿两端的核心抽象是理解整个运行逻辑的基础EventLoopGroup事件循环组包含多个NioEventLoop每个EventLoop绑定1个独立线程和1个Java NIO Selector循环处理IO事件和任务队列中的任务从根本上避免多线程同步开销。ChannelNetty对网络连接的抽象封装了底层Socket的所有IO操作所有操作都是异步执行调用后立即返回ChannelFuture对象可通过监听器异步感知操作结果。ChannelPipeline责任链模式实现的处理器流水线每个Channel绑定独立的Pipeline入站事件从头部向尾部流转出站事件从尾部向头部流转依次经过注册的ChannelHandler完成编解码、业务处理等逻辑。二、服务端完整运行流程服务端采用经典的主从Reactor双线程模型将连接监听和数据读写完全解耦流程分为5个核心阶段资源初始化与引导配置创建两个独立的EventLoopGroupbossGroup默认仅1个线程专门负责监听端口、接收客户端连接workerGroup默认线程数为CPU核心数*2负责处理所有已建立连接的IO读写事件。通过ServerBootstrap启动引导器绑定双线程组、指定NIO传输模型的NioServerSocketChannel同时配置TCP底层参数比如SO_BACKLOG设置服务端可排队的最大连接数。服务端Channel初始化与注册通过反射创建NioServerSocketChannel实例为其初始化专属的ChannelPipeline添加服务端启动过程中需要的处理器。随后将该ServerSocketChannel注册到bossGroup中的某个EventLoop的Selector上注册完成后触发通道就绪事件。端口绑定与启动完成注册成功后将NioServerSocketChannel绑定到指定的本地端口完成端口监听的初始化。此时服务端正式进入运行状态boss线程开始轮询Selector的ACCEPT事件。连接接收与分发当客户端发起TCP三次握手后Selector触发ACCEPT事件boss线程通过NioServerSocketChannel的accept方法创建对应的NioSocketChannel代表服务端侧的客户端连接随后将这个新的SocketChannel注册到workerGroup中某一个空闲的EventLoop的Selector上彻底将连接的后续处理交给worker线程池。连接的IO事件处理被分配的worker线程循环轮询Selector的读写事件当连接有数据到达时触发读事件数据经过Pipeline中的解码器、业务处理器依次处理当需要向客户端返回数据时触发写事件数据经过编码器后写入底层Socket缓冲区发送出去。同时每个EventLoop还会处理自身任务队列中的异步任务支持用户自定义提交延迟任务或普通异步任务。三、客户端完整运行流程客户端采用单Reactor线程模型所有连接、读写逻辑都由同一个EventLoopGroup处理流程分为3个核心阶段资源初始化与引导配置创建客户端专属的NioEventLoopGroup通过Bootstrap启动引导器绑定该线程组指定客户端NIO通道NioSocketChannel同时配置TCP参数比如设置连接超时时间避免长时间阻塞。通道初始化与连接建立通过ChannelInitializer为即将创建的SocketChannel初始化Pipeline按顺序添加编解码器、业务处理器遵循“解码器在前、业务处理器在中、编码器在后”的配置原则。调用connect(host, port)方法异步发起TCP连接返回ChannelFuture对象可通过添加监听器感知连接成功或失败的状态连接失败时可通过调度线程实现指数退避重连机制。通信与资源释放连接建立完成后客户端的EventLoop线程开始轮询该通道的读写事件完成和服务端的双向数据交互。当通道关闭事件触发时自动释放底层Socket资源最终优雅关闭EventLoopGroup终止所有事件循环线程。四、两端通信全链路关键细节所有IO操作全程异步没有任何阻塞等待逻辑通过ChannelListener回调机制感知操作结果大幅提升高并发场景下的吞吐量。一个EventLoop在整个生命周期内只会绑定唯一的一个线程该EventLoop下所有注册的Channel的IO事件都由这个专属线程处理天然避免了多线程并发安全问题不需要额外加锁。结合之前提到的帧拆分处理器如LengthFieldBasedFrameDecoder可以在Pipeline的最前置环节自动完成半包数据的拼接解析保证后续业务处理器拿到的永远是完整的业务消息从应用层彻底解决TCP无边界流带来的半包问题。
