关于前端:netty中channelHandler实现原理及最佳实践｜极客星球

47次阅读

共计 9671 个字符，预计需要花费 25 分钟才能阅读完成。

为继续夯实 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;
// ....

}

正文完
前端
发表至： 前端
2023-01-05
 0
关于前端:精读SolidJS
关于前端:快速构建页面结构的-3D-Visualization
关于前端:前端水印如此简单
关于前端:从零到一实现企业级微前端框架保姆级教学