从时延毛刺问题定位到 Netty 的性能统计设计

一、背景：

通常情况下，用户以黑盒的方式使用 Netty，通过 Netty 完成协议消息的读取和发送，以及编解码操作，不需要关注 Netty 的底层实现细节。

在高并发场景下，往往需要统计系统的关键性能 KPI 数据，结合日志、告警等对故障进行定位分析，如果对 Netty 的底层实现细节不了解，获取哪些关键性能数据，以及数据正确的获取方式都将成为难点。错误或者不准确的数据可能误导定位思路和方向，导致问题迟迟不能得到正确解决。

二、时延毛刺故障排查的艰辛历程:

问题现象：某电商生产环境在业务高峰期，偶现服务调用时延突刺问题，时延突然增大的服务没有固定规律，问题发生的比例虽然很低，但是对客户的体验影响很大，需要尽快定位出问题原因并解决。

• 时延毛刺问题初步分析：

服务调用时延增大，但并不是异常，因此运行日志并不会打印 ERROR 日志，单靠传统的日志无法进行有效问题定位。利用分布式消息跟踪系统，进行分布式环境的故障定界。

通过对服务调用时延进行排序和过滤，找出时延增大的服务调用链详细信息，发现业务服务端处理很快，但是消费者统计数据却显示服务端处理非常慢，调用链两端看到的数据不一致，怎么回事？

对调用链的详情进行分析发现，服务端打印的时延是业务服务接口调用的耗时，并没有包含：

（1）服务端读取请求消息、对消息做解码，以及内部消息投递、在线程池消息队列排队等待的时间。

（2）响应消息编码时间、消息队列发送排队时间以及消息写入到 Socket 发送缓冲区的时间。

服务调用链的工作原理如下：

图 1 服务调用链工作原理

将调用链中的消息调用过程详细展开，以服务端读取请求和发送响应消息为例进行说明，如下图所示：

图 2 服务端调用链详情

对于服务端的处理耗时，除了业务服务自身调用的耗时之外，还应该包含服务框架的处理时间，具体如下：

（1）请求消息的解码（反序列化）时间。

（2）请求消息在业务线程池中排队等待执行时间。

（3）响应消息编码（序列化）时间。

（4）响应消息 ByteBuf 在发送队列的排队时间。

由于服务端调用链只采集了业务服务接口的调用耗时，没有包含服务框架本身的调度和处理时间，导致无法对故障进行定界：服务端没有统计服务框架的处理时间，因此不排除问题出在消息发送队列或者业务线程池队列积压而导致时延变大。

• 服务调用链改进：

对服务调用链埋点进行优化，具体措施如下：

（1）包含客户端和服务端消息编码和解码的耗时。

（2）包含请求和应答消息在队列中的排队时间。

（3）包含应答消息在通信线程发送队列（数组）中的排队时间。

同时，为了方便问题定位，增加打印输出 Netty 的性能统计日志，主要包括：

（1）当前系统的总链路数、以及每个链路的状态。

（2）每条链路接收的总字节数、周期 T 接收的字节数、消息接收吞吐量。

（3）每条链路发送的总字节数、周期 T 发送的字节数、消息发送吞吐量。

对服务调用链优化之后，上线运行一段时间，通过分析比对 Netty 性能统计日志、调用链日志，发现双方的数据并不一致，Netty 性能统计日志统计到的数据与前端门户看到的也不一致，因此怀疑是新增的性能统计功能存在 BUG，需要继续对问题进行定位。

• 都是同步思维惹的祸:

传统的同步服务调用，发起服务调用之后，业务线程阻塞，等待响应，接收到响应之后，业务线程继续执行，对发送的消息进行累加，获取性能 KPI 数据。

使用 Netty 之后，所有的网络 I/O 操作都是异步执行的，即调用 Channel 的 write 方法，并不代表消息真正发送到 TCP 缓冲区中，如果在调用 write 方法之后就对发送的字节数做计数，统计结果就不准确。

对消息发送功能进行 code review，发现代码调用完 writeAndFlush 方法之后直接对发送的请求消息字节数进行计数，代码示例如下：

 public void channelRead(ChannelHandlerContextctx, Object msg) {        int sendBytes =((ByteBuf)msg).readableBytes();        ctx.writeAndFlush(msg);        totalSendBytes.getAndAdd(sendBytes);}

调用 writeAndFlush 并不代表消息已经发送到网络上，它仅仅是个异步的消息发送操作而已，调用 writeAndFlush 之后，Netty 会执行一系列操作，最终将消息发送到网络上，相关流程如下所示：

图 3 writeAndFlush 处理流程图

通过对 writeAndFlush 方法深入分析，我们发现性能统计代码忽略了如下几个耗时：

（1）业务 ChannelHandler 的执行时间。

（2）被异步封装的 WriteTask/WriteAndFlushTask 在 NioEventLoop 任务队列中的排队时间。

（3）ByteBuf 在 ChannelOutboundBuffer 队列中排队时间。

（4）JDK NIO 类库将 ByteBuffer 写入到网络的时间。

由于性能统计遗漏了上述 4 个关键步骤的执行时间，因此统计出来的发送速率比实际值会更高一些，这将干扰我们的问题定位思路。

• 正确的消息发送速率性能统计策略:

正确的消息发送速率性能统计方法如下：

（1）调用 writeAndFlush 方法之后获取 ChannelFuture。

（2）新增消息发送 ChannelFutureListener 并注册到 ChannelFuture 中，监听消息发送结果，如果消息写入 SocketChannel 成功，则 Netty 会回调 ChannelFutureListener 的 operationComplete 方法。

（3）在消息发送 ChannelFutureListener 的 operationComplete 方法中进行性能统计。

正确的性能统计代码示例如下：

public voidchannelRead(ChannelHandlerContext ctx, Object msg) {        int sendBytes =((ByteBuf)msg).readableBytes();        ChannelFuture writeFuture =ctx.write(msg);        writeFuture.addListener((f) ->        {           totalSendBytes.getAndAdd(sendBytes);        });        ctx.flush();}

对 Netty 消息发送相关源码进行分析，当发送的字节数大于 0 时，进行 ByteBuf 的清理工作，代码如下：

protected voiddoWrite(ChannelOutboundBuffer in) throws Exception {    //代码省略...     if (localWrittenBytes <= 0) {                        incompleteWrite(true);                        return;                    }                   adjustMaxBytesPerGatheringWrite(attemptedBytes, localWrittenBytes,maxBytesPerGatheringWrite);                    in.removeBytes(localWrittenBytes);                    --writeSpinCount;                    break;//代码省略...}

接着分析 ChannelOutboundBuffer 的 removeBytes(long writtenBytes)方法，将发送的字节数与当前 ByteBuf 可读的字节数进行对比，判断当前的 ByteBuf 是否完成发送，如果完成则调用 remove()清理它，否则只更新下发送进度，相关代码如下：

protected voiddoWrite(ChannelOutboundBuffer in) throws Exception {    //代码省略...      if (readableBytes <=writtenBytes) {                if (writtenBytes != 0) {                    progress(readableBytes);                    writtenBytes -=readableBytes;                }                remove();            } else {                if (writtenBytes != 0) {                    buf.readerIndex(readerIndex+ (int) writtenBytes);                    progress(writtenBytes);                }                break;            }//代码省略...}

当调用 remove()方法时，最终会调用消息发送 ChannelPromise 的 trySuccess 方法，通知监听 Listener 消息已经完成发送，相关代码如下所示：

public booleantrySuccess(V result) {//代码省略...        if (setSuccess0(result)) {            notifyListeners();            return true;        }        return false;    }//代码省略...}

经过以上分析可以看出，调用 write/writeAndFlush 方法本身并不代表消息已经发送完成，只有监听 write/writeAndFlush 的操作结果，在异步回调监听中计数，结果才更精确。

需要注意的是，异步回调通知由 Netty 的 NioEventLoop 线程执行，即便异步回调代码写在业务线程中，也是由 Netty 的 I/O 线程来执行累加计数的，因此这块儿需要考虑多线程并发安全问题，调用堆栈示例如下：

图 4 消息发送结果异步回调通知执行线程

如果消息报文比较大，或者一次批量发送的消息比较多，可能会出现“写半包”问题，即一个消息无法在一次 write 操作中全部完成发送，可能只发送了一半，针对此类场景，可以创建 GenericProgressiveFutureListener 用于实时监听消息发送进度，做更精准的统计，相关代码如下所示：

privatestatic void notifyProgressiveListeners0(            ProgressiveFuture<?> future,GenericProgressiveFutureListener<?>[] listeners, long progress,long total) {        for(GenericProgressiveFutureListener<?> l: listeners) {            if (l == null) {                break;            }            notifyProgressiveListener0(future,l, progress, total);        }}

问题定位出来之后，按照正确的做法对 Netty 性能统计代码进行了修正，上线之后，结合调用链日志，很快定位出了业务高峰期偶现的部分服务时延毛刺较大问题，优化业务线程池参数配置之后问题得到解决。

• 常见的消息发送性能统计误区:

在实际业务中比较常见的性能统计误区如下：

（1）调用 write/ writeAndFlush 方法之后就开始统计发送速率。

（2）消息编码时进行性能统计：编码之后，获取 out 可读的字节数，然后做累加。编码完成并不代表消息被写入到 SocketChannel 中，因此性能统计也不准确。

• Netty 关键性能指标采集:

除了消息发送速率，还有其它一些重要的指标需要采集和监控，无论是在调用链详情中展示，还是统一由运维采集、汇总和展示，这些性能指标对于故障的定界和定位帮助都很大。

• Netty I/O 线程池性能指标:

Netty I/O 线程池除了负责网络 I/O 消息的读写，还需要同时处理普通任务和定时任务，因此消息队列积压的任务个数是衡量 Netty I/O 线程池工作负载的重要指标。由于 Netty NIO 线程池采用的是一个线程池/组包含多个单线程线程池的机制，因此不需要像原生的 JDK 线程池那样统计工作线程数、最大线程数等。相关代码如下所示：

publicvoid channelActive(ChannelHandlerContext ctx) throws Exception {kpiExecutorService.scheduleAtFixedRate(()->        {            Iterator<EventExecutor>executorGroups = ctx.executor().parent().iterator();            while (executorGroups.hasNext())            {                SingleThreadEventExecutorexecutor = (SingleThreadEventExecutor)executorGroups.next();                int size = executor.pendingTasks();                if (executor == ctx.executor())                   System.out.println(ctx.channel() + "--> " + executor +" pending size in queue is : --> " + size);                else                    System.out.println(executor+ " pending size in queue is : --> " + size);            }        },0,1000, TimeUnit.MILLISECONDS);   }}

运行结果如下所示：

图 5 Netty I/O 线程池性能统计 KPI 数据

• Netty 发送队列积压消息数：

Netty 消息发送队列积压数可以反映网络速度、通信对端的读取速度、以及自身的发送速度等，因此对于服务调用时延的精细化分析对于问题定位非常有帮助，它的采集方式代码示例如下：

publicvoid channelActive(ChannelHandlerContext ctx) throws Exception {writeQueKpiExecutorService.scheduleAtFixedRate(()->        {            long pendingSize =((NioSocketChannel)ctx.channel()).unsafe().outboundBuffer().totalPendingWriteBytes();            System.out.println(ctx.channel() +"--> " + " ChannelOutboundBuffer's totalPendingWriteBytes is: "                    + pendingSize + "bytes");        },0,1000, TimeUnit.MILLISECONDS);}

执行结果如下：

图 6 Netty Channel 对应的消息发送队列性能 KPI 数据

由于 totalPendingSize 是 volatile 的，因此统计线程即便不是 Netty 的 I/O 线程，也能够正确的读取其最新值。

• Netty 消息读取速率性能统计:

针对某个 Channel 的消息读取速率性能统计，可以在解码 ChannelHandler 之前添加一个性能统计 ChannelHandler，用来对读取速率进行计数，相关代码示例如下(ServiceTraceProfileServerHandler 类)：

public voidchannelActive(ChannelHandlerContext ctx) throws Exception {       kpiExecutorService.scheduleAtFixedRate(()->        {            int readRates =totalReadBytes.getAndSet(0);            System.out.println(ctx.channel() +"--> read rates " + readRates);        },0,1000, TimeUnit.MILLISECONDS);        ctx.fireChannelActive();    }    public void channelRead(ChannelHandlerContextctx, Object msg) {        int readableBytes =((ByteBuf)msg).readableBytes();        totalReadBytes.getAndAdd(readableBytes);        ctx.fireChannelRead(msg);}

运行结果如下所示：

图 7 NettyChannel 消息读取速率性能统计

三、总结：

本文选自《Netty 进阶之路：跟着案例学 Netty》一书，由电子工业出版社出版，李林锋著。

当我们需要对服务调用时延进行精细化分析时，需要把 Netty 通信框架底层的处理耗时数据也采集走并进行分析，由于 Netty 的 I/O 操作都是异步的，因此不能像传统同步调用那样的思路去做性能数据统计，需要注册性能统计监听器，在异步回调中完成计数。另外，Netty 的 I/O 线程池、消息发送队列等实现比较特殊，与传统的 Tomcat 等框架实现策略不同，因此对于 Netty 的关键性能 KPI 数据采集不能照搬 JDK 和 Tomcat 的做法。