先抛一个场景。你维护着一个基于Netty的服务端客户端连接都建立成功了channelActive里的日志也打出来了可当你执行channel.writeAndFlush(hello)想给客户端推一条字符串消息或者用ChannelGroup广播了一圈客户端那边就是干巴巴地等着一条数据都进不来。我第一次碰上这个问题翻来覆去查服务器端口、查防火墙、甚至怀疑是不是机房网络丢包最后才发现锅在Netty自己的Pipeline和编码器身上。这篇文章就把channel、ChannelGroup、ctx.writeAndFlush()这几个东西掰开揉碎讲清楚重点说说字符串消息为什么发不出去、发出去为什么对方收不到以及从现象到根因怎么一步步定位。如果你是刚接触Netty、或者正在排查线上推送消息丢失这篇应该能帮你少走不少弯路。1. 先把三个关键对象搞清楚channel、ChannelGroup、ctx.writeAndFlush()1.1 channel是连接的抽象但不是你想的那个连接很多新手会把Netty里的Channel理解成TCP连接本身这个认知在大多数时候能用遇到问题就卡壳了。Channel本质上是一个对底层socket连接的封装它负责管理连接状态、读写缓冲区、以及向EventLoop提交IO任务。一个TCP连接对应一个Channel但Channel自己并不直接干活真正干活的是它身上挂着的Pipeline。这里有几个状态要分清。isOpen()表示Channel这个对象创建了还没被关闭哪怕底层的TCP连接已经断了只要没走close流程isOpen()可能还是true。isActive()表示连接真正处于可用状态TCP握手完成、可以正常读写。isWritable()表示当前Channel的出站缓冲区是否还能继续写入如果客户端消费速度跟不上缓冲区被写满这里就是false。排查收不到消息的时候很多人第一反应是去看服务端有没有发但往往会忽略一个前置检查这个连接到底还活着吗如果客户端早就异常断开而服务端没感知到你对着一个已经死掉的Channel写数据消息自然到不了任何地方。1.2 ChannelGroup是群发工具管理不到位就是坑ChannelGroup做的事情很朴素把一组Channel收集起来然后对它们统一执行writeAndFlush、close之类的操作。它的常用实现是DefaultChannelGroup构造时需要传一个EventExecutor一般用GlobalEventExecutor.INSTANCE就够了。管理方式很简单channelActive的时候addchannelInactive的时候remove。但ChannelGroup不是银弹。它只负责替你批量发送不负责帮你判断这个channel的pipeline能不能处理我发出去的消息类型也不负责告诉你这次广播里哪些channel其实已经废了。我见过不少人把channel塞进ChannelGroup之后就不管了连接断开也不remove最后广播的时候一个劲儿往死连接上写数据写出去的promise一个接一个失败但因为没人监听这些失败服务端日志里干净得跟什么都没发生过一样。所以用ChannelGroup一定要养成配套的习惯add和remove成对出现广播后检查返回的ChannelGroupFuture。1.3 ctx.writeAndFlush() 与 channel.writeAndFlush() 的遍历差异这俩方法看着差不多实际差别非常大是字符串消息发不出去的经典原因之一。Netty的Pipeline是一个双向链表内部维护了head和tail两个哨兵节点。读事件从head往tail方向走写事件从tail往head方向走。这里要特别注意写事件的从tail往head不是绝对的它指的是从你发起写入的位置开始往head方向找下一个能处理写事件的handler。channel.writeAndFlush(msg)是从这个Channel的Pipeline的tail开始发起所以它一定会经过链路上所有的出站处理器包括你加的各种编码器、日志处理器、自定义的outbound处理器。ctx.writeAndFlush(msg)则是从当前这个ChannelHandlerContext所在的位置开始往head方向找也就是说当前ctx后面的那些出站处理器它根本不会经过。说个具体的例子。假设你的pipeline是head - 入站解码器 - 出站编码器 - 业务handler - tail。在业务handler里ctx.channel().writeAndFlush()和ctx.writeAndFlush()都能让消息经过编码器区别不大。但如果业务handler后面还有一个自定义的outbound处理器比如统计流量、做二次封装这时候你用ctx.writeAndFlush()就会跳过它消息直接往前走了。反过来如果你在出站编码器内部调用ctx.channel().writeAndFlush()因为从tail重新发起会再次进入这个编码器造成递归栈直接爆掉。我给个对比表你写代码的时候对着看就行。方法发起位置经过哪些出站处理器适用场景ctx.writeAndFlush(msg)当前handler所在位置当前位置往head方向的所有出站处理器只想把消息写出不想被后续的处理器再做处理channel.writeAndFlush(msg)Pipeline的tail整个链路上所有出站处理器需要消息经过所有编码和自定义流程我个人的经验是在业务handler里发消息优先用ctx.writeAndFlush()因为它的行为更可控而且性能上不会多一次无谓的tail遍历。如果你确定要走完整条链路的出站处理才用channel.writeAndFlush()。2. 客户端收不到消息按顺序排查这6个原因2.1 服务端没接编码器字符串根本出不了站这是最最最常见的原因。Netty底层最终写进socket的只有ByteBuf或者FileRegion你直接写一个String进去Pipeline走到head节点附近系统发现这个消息类型没法处理会直接抛UnsupportedOperationException类似unsupported message type: String (expected: ByteBuf, FileRegion)。这个异常会让写入的ChannelFuture以失败收场。但问题在于如果你的写入没有挂任何监听器这个失败是静默的。更坑的是很多服务的handler没有重写exceptionCaught或者重写了但只打了log就完事甚至有的直接把异常方法写成了空实现。Netty默认的exceptionCaught行为是打印日志并关闭连接所以客户端那边看到的往往是连接被重置服务端这边则看到一堆异常堆栈但你要是没盯着日志看很容易漏掉。解决办法很简单在服务端的pipeline里加StringEncoderch.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8)); ch.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));StringEncoder会把CharSequence编码成ByteBufStringDecoder把ByteBuf解码成String。注意这俩handler都是有状态的天然线程安全可以使用单例但在多handler共享时要小心Sharable注解的问题。2.2 客户端的解码器和服务端的发送格式对不上服务端发出去了客户端也不代表一定能收到。客户端pipeline里必须有一个对应的解码器把ByteBuf还原成String。如果客户端只加了一个自定义handler收到的msg类型是ByteBuf你在handler里直接调msg.toString()打出来是一堆类似UnpooledHeapByteBuf(ridx: 0, widx: 15, cap: 1024)的东西看起来像没收到其实收到了但没解码。还有一种情况是分隔符问题。比如客户端使用了LineBasedFrameDecoder或者DelimiterBasedFrameDecoder期望每条消息以换行符\n或者\r\n结尾但服务端发送的字符串末尾没有带换行。这时候解码器会在内部把数据暂存起来一直等到缓冲区的数据攒够了或者遇到分隔符才算一条完整消息。你等服务端发出去hello客户端那边能收到hello\n吗收不到它还在傻等那个换行符。这种问题在Netty里有个学名叫粘包拆包属于TCP流式传输的特点你发的多条消息在网络上根本没有边界必须靠协议层自己划界。如果你的场景是发不带分隔符的字符串建议改成LengthFieldBasedFrameDecoder前面加一个长度字段这样不会依赖分隔符但实现复杂度会高一些。2.3 只write没flush消息在缓冲区里躺着等Netty 4.x之后write和flush是分离的。write只是把消息放进出站缓冲区也就是ChannelOutboundBuffer并没有真正写到socket上。flush才是把缓冲区的数据刷出去。只调write不调flush数据可能一直积压着客户端自然收不到。很多人是从Netty 3.x的老代码迁移上来的那时候write之后有自动flush或者默认行为不一样导致这个坑在新版本特别容易踩。我的建议很简单能用writeAndFlush就用writeAndFlush别再分开写。如果你确实要分开注意flush的粒度高频写入场景可以批量write然后一次性flush但一定要保证最终会flush一次否则线上会出间歇性消息延迟。顺便提一句TCP_NODELAY。如果没开启这个选项Nagle算法会把多个小包合并发送配合客户端的Delayed ACK小消息可能被延迟几十毫秒甚至更久。虽然不至于完全收不到但表现出来就是有时候要等好几秒。开发环境测试字符串消息直接两端都加上.option(ChannelOption.TCP_NODELAY, true) .childOption(ChannelOption.TCP_NODELAY, true)2.4 连接还没激活就写数据写入静默失败这个坑我在写客户端的时候踩过。Bootstrap.connect()返回的ChannelFuture是连接发起的结果不是连接建立完成的结果。如果拿到Channel之后立刻writeAndFlushTCP三次握手可能还没完成消息就发出去了。这种情况下写入可能以NotYetConnectedException或者IllegalStateException失败而且大多数时候这些异常不会出现在你的业务代码里因为没有promise监听。等到你回头排查看到的就是客户端那边风平浪静服务端什么都没收到。正确做法有两种。一种是给connect的future挂监听bootstrap.connect(host, port).addListener((ChannelFuture f) - { if (f.isSuccess()) { f.channel().writeAndFlush(hello); } });另一种更推荐把初始化发送动作放在handler的channelActive回调里这个回调触发的时机就是连接可用的时候。写客户端代码时凡是涉及连接成功后的首次写入都要检查发起时机。2.5 ChannelGroup广播时目标channel已经被移除了广播场景下最经典的问题就是往一个已经关闭的channel上写数据。客户端断网、超时、主动关闭服务端如果没有及时感知这个channel还会留在ChannelGroup里。每次广播它都会收到一份写入任务然后写入失败promise异常但因为没人监听服务端日志干净得像什么都没发生。这种情况的表现很有迷惑性一部分客户端能收到消息另一部分收不到而且收不到的那部分每次都不一样因为跟你广播的时机有关。排查的时候你会发现服务端确实在发但为什么某些客户端收不到你再看一眼连接状态就明白了那些客户端早就离线了只是ChannelGroup里还保着它们的僵尸连接。修复方案两个动作缺一不可。第一在channelInactive里removeOverride public void channelInactive(ChannelHandlerContext ctx) { CLIENTS.remove(ctx.channel()); }第二广播之后不要什么都不做要检查ChannelGroupFutureChannelGroupFuture future CLIENTS.writeAndFlush(message); future.addListener(f - { if (!f.isSuccess()) { future.forEach(cf - { if (!cf.isSuccess()) { System.out.println(发送失败 channel cf.channel() cause cf.cause()); } }); } });2.6 事件循环线程被阻塞写入任务排队到地老天荒还有一个不那么明显的原因是EventLoop被长时间阻塞。EventLoop是Netty处理读写事件的核心线程它同时服务多个channel。如果你在某个channel的handler里做了耗时操作比如数据库查询、外部HTTP调用、Thread.sleep那么这个EventLoop上的所有channel都会被拖慢。写入任务虽然通过ctx.writeAndFlush()提交了但EventLoop一直忙着执行前面的阻塞任务写入迟迟不被处理客户端自然也就收不到消息。这种问题的特征是服务端看起来没报错但整体吞吐下降所有连接都延迟。排查方法很简单遇到疑似情况先抓线程栈看看EventLoop线程在干什么。生产环境里的规矩是任何耗时操作都丢给业务线程池不要在EventLoop里做。3. 一次真实排障实录从收不到到根因修复3.1 现场情况能连上、能读唯独不能写我之前维护过一个在线列表的推送服务服务端维护了一个ChannelGroup每5秒广播一次在线人数。上线一段时间后运营反馈有些客户端的在线人数再也不刷新了。检查客户端日志显示连接一直活着也没有明显的报错。服务端看监控广播任务一直在跑日志里也打了广播完成。最诡异的是同一批客户端里一部分能正常收到另一部分收不到而且收不到的那批是逐渐变多的。第一反应是网络问题但抓包之后发现服务端和客户端之间的TCP连接在某些客户端上根本没有活跃的数据包。这就奇怪了客户端明明显示连接正常。3.2 完整排查链路抓包、看Pipeline、看日志我先把服务端广播代码从头到尾看了一遍发现问题出在ChannelGroup的管理上。服务端的channelInactive没有被重写也就是说连接断开后没有任何代码把channel从ChannelGroup里remove出去。那些收不到的客户端客户端进程早就因为网络波动、移动端切后台等原因断开了但服务端不知道。为什么服务端不知道TCP连接异常断开时服务端要依赖TCP的KeepAlive或者自己实现的心跳来感知。我们当时只做了业务层面的5秒广播没有做双向心跳客户端异常断电后服务端要等很久才能通过一次写入失败感知到连接已死。而在这个感知发生之前ChannelGroup里全是这种僵尸连接。这时候再看广播逻辑每次广播CLIENTS.writeAndFlush()会对所有channel发起写入。对已关闭的channel写入会以ClosedChannelException失败。但我们的广播代码只调了writeAndFlush没有挂监听器这个失败在网上悄无声息地消失了。服务端的日志当然干净。3.3 根因与修复promise失败为什么没人知道根因一句话总结不是消息没发出去而是发给了已经死了的连接且没人检查发送结果。修复方案有三步。第一步在channelInactive里remove连接。第二步广播后遍历ChannelGroupFuture排查失败项至少把失败原因打印出来。第三步加上心跳机制让服务端能及时感知死连接而不是靠广播的时候撞运气。这个案例给我最大的教训是writeAndFlush的返回值是有意义的。Netty里几乎所有IO操作都是异步的返回值是一个Future它承载了这次操作成功还是失败的最终结果。你不对它负责它就默认帮你把错误吞掉然后让你的程序在一种看似正常但实际已经出问题的状态里跑很久。后来我们团队定了个规矩线上代码里所有writeAndFlush返回的future必须挂至少一个监听器哪怕是打一行日志。4. 可直接抄作业的字符串收发方案4.1 服务端完整代码单发群发直接给一份能跑起来的服务端代码pipeline、编码器、ChannelGroup管理都放在一起你照着改端口和业务逻辑就能用。public class StringServer { private static final ChannelGroup CLIENTS new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); public static void main(String[] args) throws Exception { EventLoopGroup boss new NioEventLoopGroup(1); EventLoopGroup worker new NioEventLoopGroup(); try { ServerBootstrap bootstrap new ServerBootstrap(); bootstrap.group(boss, worker) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childOption(ChannelOption.TCP_NODELAY, true) .childHandler(new ChannelInitializerSocketChannel() { Override protected void initChannel(SocketChannel ch) { ch.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8)); ch.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8)); ch.pipeline().addLast(new ServerHandler()); } }); ChannelFuture bindFuture bootstrap.bind(8080).sync(); System.out.println(server started on 8080); bindFuture.channel().closeFuture().sync(); } finally { boss.shutdownGracefully(); worker.shutdownGracefully(); } } }ServerHandler负责连接管理和消息处理。注意在channelActive里addchannelInactive里remove改header异常处理一定不要空实现。public class ServerHandler extends ChannelInboundHandlerAdapter { Override public void channelActive(ChannelHandlerContext ctx) { CLIENTS.add(ctx.channel()); System.out.println(online: ctx.channel().remoteAddress()); // 单发用ctx写只走当前handler之前的出站处理器 ctx.writeAndFlush(welcome\n) .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); } Override public void channelInactive(ChannelHandlerContext ctx) { CLIENTS.remove(ctx.channel()); System.out.println(offline: ctx.channel().remoteAddress()); } Override public void channelRead(ChannelHandlerContext ctx, Object msg) { String text (String) msg; System.out.println(recv from ctx.channel().remoteAddress() : text); // 群发给所有在线连接广播 ChannelGroupFuture broadcastFuture CLIENTS.writeAndFlush([broadcast] text \n); broadcastFuture.addListener(f - { if (!f.isSuccess()) { broadcastFuture.forEach(cf - { if (!cf.isSuccess()) { System.err.println(broadcast fail: cf.channel() - cf.cause()); } }); } }); } Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); } }4.2 客户端完整代码解码与输出客户端同样要配好解码器和编码器。收到消息后因为pipeline里有StringDecoderchannelRead0里拿到的msg直接就是String不用再手动转ByteBuf。public class StringClient { public static void main(String[] args) throws Exception { EventLoopGroup group new NioEventLoopGroup(); try { Bootstrap bootstrap new Bootstrap(); bootstrap.group(group) .channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) .handler(new ChannelInitializerSocketChannel() { Override protected void initChannel(SocketChannel ch) { ch.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8)); ch.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8)); ch.pipeline().addLast(new SimpleChannelInboundHandlerString() { Override protected void channelRead0(ChannelHandlerContext ctx, String msg) { System.out.println(recv: msg); } }); } }); Channel channel bootstrap.connect(127.0.0.1, 8080).sync().channel(); channel.writeAndFlush(hello server\n) .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); channel.closeFuture().sync(); } finally { group.shutdownGracefully(); } } }4.3 生产环境的几个加固点上面的代码能跑通本地测试但放到生产环境还需要注意几个点。不要在main线程里用while循环发送广播应该用EventLoop的schedule方法。这样能保证广播任务和IO事件在同一个线程里串行执行避免多线程同时操作Channel的竞态问题。// 在某个handler的channelActive里启动定时任务 ctx.executor().scheduleAtFixedRate(() - { CLIENTS.writeAndFlush(heartbeat\n) .addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE); }, 5, 5, TimeUnit.SECONDS);如果你要发大量消息先用channel.isWritable()做背压判断否则客户端消费不过来缓冲区会越积越大最后内存出问题。开发阶段可以在pipeline最前面加一个LoggingHandler能直接看到每条读写消息的类型和内容这个工具在排查到底发没发出去的时候特别好使。5. 常见问题速查表与我的排错习惯5.1 症状、原因、处理对照速查症状常见原因处理方式服务端日志出现unsupported message type: Stringpipeline缺StringEncoder加上StringEncoder(UTF_8)客户端handler收到的是ByteBuf客户端缺StringDecoder加上StringDecoder客户端用LineBasedFrameDecoder但一直没消息发送的字符串不带换行符消息加\n或换长度字段协议客户端很久才收到一次消息未开启TCP_NODELAY或只write没flush两端开TCP_NODELAY统一writeAndFlushconnect后立刻write抛IllegalStateExceptionchannel还没注册到EventLoop在channelActive里发或给connect挂监听部分客户端收不到广播无任何报错ChannelGroup里有已关闭连接且没监听FuturechannelInactive里remove检查ChannelGroupFuture客户端收到乱码服务端和客户端字符集不一致统一使用UTF_8所有连接延迟严重服务端线程看着卡住EventLoop被耗时操作阻塞耗时操作丢业务线程池写入时出现ClosedChannelException往已关闭的连接写数据写前判isActive写后监听Future5.2 三个踩坑后养成的编码习惯先说检查Future。所有writeAndFlush调用线上代码一律挂监听器。最简单粗暴的写法是addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE)它不仅帮你打印异常还会把异常重新抛给这个Channel的exceptionCaught链路这样至少不会静默丢失。我见过太多线上事故根源都是那个写失败的异常没人看。再说画Pipeline。我每次写Netty的handler之前都会先在纸上或者注释里画出Pipeline的结构标清楚哪个是inbound、哪个是outbound然后问自己一个问题我这条消息从发起到真正写进socket会经过哪些handler尤其是用ctx.writeAndFlush()还是channel.writeAndFlush()画完图之后一眼就能确定。别嫌麻烦这个动作能避免一半以上的消息丢失问题。最后说异常处理。exceptionCaught不要写空实现不要只打一行debug日志。哪怕你暂时不知道怎么处理至少要把堆栈打印出来。很多类似的消息收不到问题其实服务端早就在exceptionCaught里暴露过根因只是没人看。等到线上出问题再回来翻日志那感觉是真的酸爽。顺便再提一个我自己现在写代码的习惯开发环境必加LoggingHandler测试完再摘掉。Netty的LoggingHandler会打印每条入站、出站消息的类名、字节长度和内容摘要排查客户端收不到的时候它能直接告诉你消息到底有没有走到对应的节点。有一次我花了一下午定位一个广播问题最后发现是有人在handler里做了消息体拦截直接把字符串转成了别的东西。有了LoggingHandler这种事情基本十分钟就能定位。