为持续夯实MobTech袤博科技的数智技术创新能力和技术布道能力，本期极客星球邀请了企业服务研发部工程师梁立从 TCP 的粘包/半包、 Netty 处理粘包/半包及源码分析、 开源项目对 channelHandler最佳实践三方面对《netty 中channelHandler的原理与最佳实践》进行了全面的技术分享。
版本信息本次分析版本基于netty 4.1.40.Final
TCP 的粘包/半包问题
在TCP/IP 协议传输网络数据包时，用户发送消息ABCD，服务端可能收到是ABCD. AB？CD？等。对于粘包问题，主要原因是发送方每次写入数据小于套接字缓冲区大小， 以及接受方读取消息不及时。对于半包问题， 主要原因是发送方每次写入数据大于套接字缓冲区大小，以及发送数据大于协议最大传输单位，底层需要拆包。那么针对此类问题，应当如何解决呢 ？常见的方式解码方式有三种：固定长度，使用固定分隔符来分割消息，以及固网长度字段存放内容长度信息。
解码实现思考
在分析之前，我们可以思考一下，如果是我们来实现上面三种编解码会如何实现 ？
我们可以整理如下需求：
1.我们需要存放我们解码好的消息；
2.我们需要提供一个解码方法来让不同子类实现， 例如固定长度，分隔符，以及固定长度字段解码的方式肯定有差别；
3.我们从套接字读取消息后就可以让我们解码器去处理了。
针对上述需求，我们还需要带着三个问题，查看源码看下是否和我们猜想的类似：
问题1：我们需要一个集合存放我们解码的消息；
问题2：我们需要不同子类对解码细节做不同实现，所以我们需要有一个父类；ByteToMessageDecoder， 可以在父类实现公共逻辑，提供给子类一个decode(List out,ByteBuf in); 方法；
问题3 ：我们从套接字读取数据之后，发送一个读事件(fireChannelRead)让我们解码器去处理。
Netty 处理粘包/半包及源码分析
封帧方式 |
解码 |
固定长度 |
FixedLengthFrameDecoder |
分隔符 |
DelimiterBasedFrameDecoder |
固定长度字段存内容长度信息 |
LengthFieldBasedFrameDecoder |
我们以固定长度解码器为例：
ServerBootstrap b = new ServerBootstrap();
// ....
b..childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new FixedLengthFrameDecoder(2));
//.... 后续业务处理handler
}
});
public class FixedLengthFrameDecoder extends ByteToMessageDecoder {
//....
}
public class ByteToMessageDecoder {
// ....
protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
}
我们查看 FixedLengthFrameDecoder ，发现果然继承父类ByteToMessageDecoder，然后父类也有一个channelRead方法处理消息，并提供一个decode抽象方法让子类实现。
channelRead
假设我们发送端发送ABCD消息，从套节字读取之后，后续会调用channelRead 方法进行解码。
我们看到获取一个集合实例CodecOutputList， 该类实现List接口。如果是首次调用，会把当前ByteBuf 赋值给cumulation，并调用callDecode(ctx, cumulation, out)。
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof ByteBuf) {
CodecOutputList out = CodecOutputList.newInstance();
try {
ByteBuf data = (ByteBuf) msg;
first = cumulation == null;
if (first) {
cumulation = data;
} else {
cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
}
callDecode(ctx, cumulation, out);
} catch (DecoderException e) {
throw e;
} catch (Exception e) {
throw new DecoderException(e);
} finally {
//.....
}
} else {
ctx.fireChannelRead(msg);
}
}
callDecode
通过字面意思就知道这个方法会做和解码相关操作。首先会判断in.isReadable() 是否可读，然后我们的outSize 目前是空， 进入到 decodeRemovalReentryProtection , 该方法会调用子类FixedLengthFrameDecoder的decode方法进行具体解码，该decode 方法比较简单就是当从ByteBuf 读取到指定长度就添加到out 中。我们读取完成后， outSize == out.size() 和 oldInputLength == in.readableBytes()都不满足，进入下一次循环， 我们outSize 大于0， 发送fireChannelRead。到此消息就被解码，并发送给我们业务channelHandler 。
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
try {
while (in.isReadable()) {
int outSize = out.size();
if (outSize > 0) {
fireChannelRead(ctx, out, outSize);
out.clear();
// Check if this handler was removed before continuing with decoding.
// If it was removed, it is not safe to continue to operate on the buffer.
//
// See:
// - https://github.com/netty/netty/issues/4635
if (ctx.isRemoved()) {
break;
}
outSize = 0;
}
int oldInputLength = in.readableBytes();
//decode中时，不能执行完handler remove清理操作。
//那decode完之后需要清理数据。
decodeRemovalReentryProtection(ctx, in, out);
// Check if this handler was removed before continuing the loop.
// If it was removed, it is not safe to continue to operate on the buffer.
//
// See https://github.com/netty/netty/issues/1664
if (ctx.isRemoved()) {
break;
}
if (outSize == out.size()) {
if (oldInputLength == in.readableBytes()) {
break;
} else {
continue;
}
}
if (oldInputLength == in.readableBytes()) {
throw new DecoderException(
StringUtil.simpleClassName(getClass()) +
".decode() did not read anything but decoded a message.");
}
if (isSingleDecode()) {
break;
}
}
} catch (DecoderException e) {
throw e;
} catch (Exception cause) {
throw new DecoderException(cause);
}
}
final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
throws Exception {
decodeState = STATE_CALLING_CHILD_DECODE;
try {
decode(ctx, in, out);
} finally {
boolean removePending = decodeState == STATE_HANDLER_REMOVED_PENDING;
decodeState = STATE_INIT;
if (removePending) {
handlerRemoved(ctx);
}
}
}
public class FixedLengthFrameDecoder extends ByteToMessageDecoder {
@Override
protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
Object decoded = decode(ctx, in);
if (decoded != null) {
out.add(decoded);
}
}
protected Object decode(
@SuppressWarnings("UnusedParameters") ChannelHandlerContext ctx, ByteBuf in) throws Exception {
if (in.readableBytes() < frameLength) {
return null;
} else {
return in.readRetainedSlice(frameLength);
}
}
}
channelHandler 的最佳实践
了解Netty 的小伙伴都知道channelHandler 分为ChannelInboundHandler 和 ChannelOutboundHandler， 分别用来处理inbound 和 outbound。
channelHandler 的最佳实践本质就是inbound 和outbound 的最佳实践。
下面列举了三种具有代表性的场景
• 按照职责划分channelHandler，例如有处理编解码，有处理心跳的，有专门处理业务的；
• 因为channel和eventLoop 线程绑定，然后一个evnetLoop 可能服务多个channel，所以我们不要在channelHandler 做耗时操作；
• outbound 我们可以优化写，减少系统调用。
按照职责划分channelHandler
rocketMq
我们可以查看rocketMq 是如何划分channelHandler ， 比如具有专门处理编/解码的NettyEncoder/NettyDecoder，通过IdleStatHandler 发现不活跃连接，管理连接handlerNettyConnectManageHandler 进行处理，
业务处理 NettyServerHandler 。
dubbo
处理编解码，检查不活跃channel，以及业务处理handler。
不在channelHandler 做耗时操作
之前介绍过一个eventLoop 线程服务多个channel，假设某个channelHandler处理耗时的任务，会影响其他channel，所以我们不要在channelHandler 执行耗时操作。
如果确实需要执行耗时操作，我们可以给channelHandler 添加一个线程池处理
final DefaultEventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup();
// 为我们的serverHandler 添加单独的线程池处理事件。
pipeline.addLast(defaultEventLoopGroup,serverHandler)；
outbound 优化写
writeAndFlush存在的问题
我们来看一下下面代码有什么问题？
public class EchoServerHandler
extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ctx.writeAndFlush(msg);
}
}
代码的问题在于ctx.writeAndFlush 每次调用都会触发一次系统调用。然后channelRead 在一次业务处理中可能被调用多次，问题就变为一次业务请求，执行多次系统调用。
优化writeAndFlush
怎么优化？
我们可以重写channelRead 和 channelReadComplete，在channelRead 中调用write 方法，
在channelReadComplete中调用flush 方法 。
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ctx.write(msg);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
}上面的实现方式确实减少系统调用，但是在netty 内部当有数据可读，会默认会连续16次，最后在调用channelReadComplete() 方法。
默认的行为存在两个问题：
1.写出数据到对端的时间被延迟了；
2.默认16 次这个数据不一定适合所有业务场景（不够灵活）。
我们需要结合业务的特性，例如业务如果关注吞吐量，可以适当把读取几次后刷新设置的大一些。如果业务关注及时性，读取几次后刷新就适当设置小一点。基于上述需求，FlushConsolidationHandler 就诞生了， 可以指定读取几次后刷新一次。
FlushConsolidationHandler 优化写
使用在pipeline中添加FlushConsolidationHandler，读取几次刷新一次可以根据业务设置，例如这里设置5次，我们是优化 EchoServerHandler的写，就放在它的前面。
// 每5次就触发一次flush
// ....
p.addLast(new FlushConsolidationHandler(5));
p.addLast(new EchoServerHandler());
// ....
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ctx.writeAndFlush(msg);
}
}
原理分析：
首先FlushConsolidationHandler 继承 ChannelDuplexHandler，能同时处理入站和出站消息，
入站我们查看 channelRead 和 channelReadComplete 实现，出站我们查看 flush 方法 （没有对write方法进行重写）。
channelRead
• 设置readInProgress 就把事件向下传递
• 我们的EchoServerHandler 会channelRead 会被调用,我们在channelRead 中调用ctx.writeAndFlush。
• 触发write 和 flush 的出站消息， FlushConsolidationHandler的flush进行处理
• 先判断readInProgress， ++flushPendingCount == explicitFlushAfterFlushes 判断是否达到期望刷新次数，我们设置为5 ，不执行刷新。
• 接着channelReadComplete 被调用，会重置准备刷新次数，并执行刷新。
关键就在channelRead 和 channelReadComplete
假设我们channelRead 读取了多次， 当读取次数大于等于5次就会刷新，小于5次时由channelReadComplete 刷新。
这样就达到了减少系统调用并且每读取几次在刷新也可以配置
public class FlushConsolidationHandler extends ChannelDuplexHandler {
// explicitFlushAfterFlushes 表示几次flush后，才真正调用flush 方法
// consolidateWhenNoReadInProgress 支持异步的情况，当readInProgress不为true 也可以支持flush
public FlushConsolidationHandler(int explicitFlushAfterFlushes, boolean consolidateWhenNoReadInProgress){
//....
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
readInProgress = true;
ctx.fireChannelRead(msg);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
// This may be the last event in the read loop, so flush now!
// 内部就是将 readInProgress = false; 当flushPendingCount 就调用flush
resetReadAndFlushIfNeeded(ctx);
ctx.fireChannelReadComplete();
}
@Override
public void flush(ChannelHandlerContext ctx) throws Exception {
//根据业务线程是否复用IO线程两种情况来考虑：
//复用情况
if (readInProgress) { //正在读的时候
// If there is still a read in progress we are sure we will see a channelReadComplete(...) call. Thus
// we only need to flush if we reach the explicitFlushAfterFlushes limit.
//每explicitFlushAfterFlushes个“批量”写（flush）一次
//不足怎么办？channelReadComplete会flush掉后面的
if (++flushPendingCount == explicitFlushAfterFlushes) {
flushNow(ctx);
}
//以下是非复用情况：异步情况
} else if (consolidateWhenNoReadInProgress) {
//（业务异步化情况下）开启consolidateWhenNoReadInProgress时，优化flush
//（比如没有读请求了，但是内部还是忙的团团转，没有消化的时候，所以还是会写响应）
// Flush immediately if we reach the threshold, otherwise schedule
if (++flushPendingCount == explicitFlushAfterFlushes) {
flushNow(ctx);
} else {
scheduleFlush(ctx);
}
} else {
//（业务异步化情况下）没有开启consolidateWhenNoReadInProgress时，直接flush
// Always flush directly
flushNow(ctx);
}
}
}
附录
默认读取16次设置入口源码分析
默认创建DefaultChannelConfig ，会接着调用重载的构造函数。
在setRecvByteBufAllocator可以看到获取metadata.defaultMaxMessagesPerRead()。
而ChannelMetadata 默认构造为 16次 new ChannelMetadata(false, 16)。
public abstract class AbstractNioByteChannel extends AbstractNioChannel {
private static final ChannelMetadata METADATA = new ChannelMetadata(false, 16);
//.....
}
// 默认选择自适应接受缓存分配器,然后在调用setRecvByteBufAllocator。
// setRecvByteBufAllocator就是指定最大读取多少次的入口 ，默认为16次
public class DefaultChannelConfig implements ChannelConfig {
public DefaultChannelConfig(Channel channel) {
//除UDP外都默认选择自适应接受缓存分配器
this(channel, new AdaptiveRecvByteBufAllocator());
}
protected DefaultChannelConfig(Channel channel, RecvByteBufAllocator allocator) {
//UDP的使用固定SIZE的接受缓存分配器：FixedRecvByteBufAllocator
setRecvByteBufAllocator(allocator, channel.metadata());
this.channel = channel;
}
}
private void setRecvByteBufAllocator(RecvByteBufAllocator allocator, ChannelMetadata metadata) {
if (allocator instanceof MaxMessagesRecvByteBufAllocator) {
((MaxMessagesRecvByteBufAllocator) allocator).maxMessagesPerRead(metadata.defaultMaxMessagesPerRead());
} else if (allocator == null) {
throw new NullPointerException("allocator");
}
setRecvByteBufAllocator(allocator);
}
public final class ChannelMetadata {
private final boolean hasDisconnect;
private final int defaultMaxMessagesPerRead;
// ....
} 