Netty学习整理

作者: 愤怒的老照 | 来源:发表于2021-04-07 21:00 被阅读0次

1 复习NIO

https://www.jianshu.com/p/b36b4e8c1343

2 Netty整体架构

image.png

2.1 网络通信层

网络通信层的职责是执行网络 I/O 的操作。它支持多种网络协议和 I/O 模型的连接操作。当网络数据读取到内核缓冲区后，会触发各种网络事件，这些网络事件会分发给事件调度层进行处理。

网络通信层的核心组件包含BootStrap&ServerBootStrap、Channel三个组件。

BootStrap（客户端） & ServerBootStrap（服务端）：负责整个 Netty 程序的启动、初始化、服务器连接等过程，它相当于一条主线，串联了 Netty 的其他核心组件。
Channel：Channel提供了基本的 API 用于网络 I/O 操作，提供了与底层 Socket 交互的能力，如 register、bind、connect、read、write、flush 等。当然 Channel 会有多种状态，如连接建立、连接注册、数据读写、连接销毁等。随着状态的变化，Channel 处于不同的生命周期，每一种状态都会绑定相应的事件回调：

image.png

2.2 事件调度层

事件调度层的职责是通过 Reactor 线程模型对各类事件进行聚合处理，通过 Selector 主循环线程集成多种事件（ I/O 事件、信号事件、定时事件等），实际的业务处理逻辑是交由服务编排层中相关的 Handler 完成。事件调度层的核心组件包括 EventLoopGroup、EventLoop。

EventLoopGroup ：本质是一个线程池，主要负责接收 I/O 请求，并分配线程执行处理请求
EventLoop：同一时间会与一个线程绑定，每个 EventLoop 负责处理多个 Channel。

EventLoopGroup与线程模型的对应？
Netty 通过创建不同的 EventLoopGroup 参数配置，就可以支持 Reactor 的三种线程模型：

单线程模型：EventLoopGroup 只包含一个 EventLoop，Boss 和 Worker 使用同一个EventLoopGroup；
多线程模型：EventLoopGroup 包含多个 EventLoop，Boss 和 Worker 使用同一个EventLoopGroup；
主从多线程模型：EventLoopGroup 包含多个 EventLoop，Boss 是主 Reactor，Worker 是从 Reactor，它们分别使用不同的 EventLoopGroup，主 Reactor 负责新的网络连接 Channel 创建，然后把 Channel 注册到从 Reactor。

2.3 服务编排层

服务编排层的职责是负责组装各类服务，它是 Netty 的核心处理链，用以实现网络事件的动态编排和有序传播。

服务编排层的核心组件包括 ChannelPipeline、ChannelHandler、ChannelHandlerContext。

ChannelPipeline：负责组装各种 ChannelHandler，当 I/O 读写事件触发时，ChannelPipeline 会依次调用 ChannelHandler 列表对 Channel 的数据进行拦截和处理。
ChannelHandler & ChannelHandlerContext：ChannelHandlerContext 用于保存 ChannelHandler 上下文，通过 ChannelHandlerContext 我们可以知道 ChannelPipeline 和 ChannelHandler 的关联关系

3 服务启动流程

public static void main(String[] args) throws Exception {
        // Configure SSL.
        final SslContext sslCtx;
        if (SSL) {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
        } else {
            sslCtx = null;
        }

        // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final EchoServerHandler serverHandler = new EchoServerHandler();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 100)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     if (sslCtx != null) {
                         p.addLast(sslCtx.newHandler(ch.alloc()));
                     }
                     //p.addLast(new LoggingHandler(LogLevel.INFO));
                     p.addLast(serverHandler);
                 }
             });

            // Start the server.
            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

配置线程池：当前配置采用主从Reactor模式。Boss 是主 Reactor，Worker 是从 Reactor。它们分别使用不同的 NioEventLoopGroup，主 Reactor 负责处理 Accept，然后把 Channel 注册到从 Reactor 上，从 Reactor 主要负责 Channel 生命周期内的所有 I/O 事件。

如果是使用group(EventLoopGroup group)方法，最终调用逻辑如下：

/**
     * Specify the {@link EventLoopGroup} which is used for the parent (acceptor) and the child (client).
     */
    @Override
    public ServerBootstrap group(EventLoopGroup group) {
        return group(group, group);
    }

可以看出Netty 线程模型的可定制化程度很高，它只需要简单配置不同的参数，便可启用不同的 Reactor 线程模型
在注册channel时，group通过chooser（有GenericEventExecutorChooser和PowerOfTwoEventExecutorChooser两种）选取eventLoop，将channel和eventLoop绑定，之后 Channel 生命周期内的所有 I/O 事件都由这个 EventLoop 负责处理

ServerBootstrap 中 bind() 实现：

创建channel

 final ChannelFuture regFuture = initAndRegister();

// initAndRegister实现
// 1、通过反射创建服务端Channel
channel = channelFactory.newChannel();

// 通过工厂类ReflectiveChannelFactory反射创建channel，类型是最开始设置的（NioServerSocketChannel）
public NioServerSocketChannel() {
    // 创建服务端ServerSocketChannel
    this(newSocket(DEFAULT_SELECTOR_PROVIDER)); 
}

// 在创建Channel时，注册感兴趣的事件（OP_ACCEPT）
public NioServerSocketChannel(ServerSocketChannel channel) {
    super(null, channel, SelectionKey.OP_ACCEPT); // 调用父类方法
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

// Channel父类构造函数
// 1、调用 AbstractChannel 的构造函数创建了三个重要的成员变量，分别为 id、unsafe、pipeline。id 表示全局唯一的 Channel，unsafe 用于操作底层数据的读写操作，pipeline 负责业务处理器的编排。
// 2、可以看到每创建一个channel，都有与之对应的pipeline，pipeline最初包含headContext和TailContext两个节点
protected AbstractChannel(Channel parent) {
    this.parent = parent;
    id = newId(); // Channel 全局唯一 id 
    unsafe = newUnsafe(); // unsafe 操作底层读写
    pipeline = newChannelPipeline(); // pipeline 负责业务处理器编排
}

private static ServerSocketChannel newSocket(SelectorProvider provider) {
    try {
        // 创建 JDK 底层的 ServerSocketChannel
        return provider.openServerSocketChannel(); 
    } catch (IOException e) {
        throw new ChannelException(
                "Failed to open a server socket.", e);
    }
}

初始化channel

 final ChannelFuture regFuture = initAndRegister();

 init(channel);

// 1、添加特殊的 Handler 处理器ChannelInitializer，并引入另一个特殊的Handler处理器ServerBootstrapAcceptor
// ChannelInitializer：提供了一个简单的工具，用于在某个Channel注册到EventLoop后，对这个Channel执行一些初始化操作，但是在初始化完成之后，ChannelInitializer会将自己从pipeline中移除，不会影响后续的操作。

//  ServerBootstrapAcceptor：成功构造客户端 NioSocketChannel 后， pipeline.fireChannelRead() 触发 channelRead 事件传播。会传播到 ServerBootstrapAcceptor.channelRead() 方法，channelRead() 会将客户端 Channel 分配到工作线程组中去执行。

p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(final Channel ch) throws Exception {
                final ChannelPipeline pipeline = ch.pipeline();
                ChannelHandler handler = config.handler();
                if (handler != null) {
                    pipeline.addLast(handler);
                }

                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        pipeline.addLast(new ServerBootstrapAcceptor(
                                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                    }
                });
            }
        });

注册channel

ChannelFuture regFuture = config().group().register(channel);

// 我们初始化的group通常是MultithreadEventLoopGroup类型的，当注册时会使用选择器选择一个eventLoop进行注册
public ChannelFuture register(Channel channel) {
    return next().register(channel); // 选择一个 eventLoop 注册
}

// 调用 JDK 底层的 register() 进行注册
doRegister(); 

//调用 JDK 底层的 register() 进行注册。register() 的第三个入参传入的是 Netty 自己实现的 Channel 对象，调用 register() 方法会将它绑定在 JDK 底层 Channel 的 attachment 上。这样在每次 Selector 对象进行事件循环时，Netty 都可以从返回的 JDK 底层 Channel 中获得自己的 Channel 对象。
selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);

注册完成后的事件触发（涉及pipeline如何和自定义的Handler关联）

// 1、用户自定义的业务处理器添加到 Pipeline 中
pipeline.invokeHandlerAddedIfNeeded();

// 通过invokeHandlerAddedIfNeeded调用Handler的handlerAdded实现，此处和之前创建的ChannelInitializer关联起来了
private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
       if (initMap.add(ctx)) {
        try {
       //  调用ChannelInitializer实现类的具体实现
       // 1、在服务端Handler：对应的是添加ServerBootstrapAcceptor这个特殊的Handler，并且在处理完之后会删除ChannelInitializer，此时Handler责任链是HeadContext<->LogHandler<->ServerBootstrapAcceptor<->TailContext
       // 2、在客户端Handler：对应的就是EchoServer添加的自定义Handler
            initChannel((C) ctx.channel()); 
        } catch (Throwable cause) {
            exceptionCaught(ctx, cause);
        } finally {
            ChannelPipeline pipeline = ctx.pipeline();
            if (pipeline.context(this) != null) {
                // 处理完之后会删除ChannelInitializer
                pipeline.remove(this); 
            }
        }
        return true;
    }
    return false;
    }

// 2、触发channelRegistered事件
pipeline.fireChannelRegistered();

端口绑定

// ...省略其他代码
javaChannel().bind(localAddress, config.getBacklog());

// 完成端口绑定之后，Channel 处于活跃 Active 状态，然后会调用 pipeline.fireChannelActive() 方法触发 channelActive 事件
pipeline.fireChannelActive(); // 触发 channelActive 事件

4 处理请求流程

NioEventLoop核心处理逻辑

protected void run() {
        // 1、入口死循环，不断检测IO事件并处理任务
        for (;;) {
            try {
                switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                    case SelectStrategy.CONTINUE:
                        continue;

                    case SelectStrategy.BUSY_WAIT:

                    case SelectStrategy.SELECT:
                        select(wakenUp.getAndSet(false));

                        

                        if (wakenUp.get()) {
                            selector.wakeup();
                        }
                    default:
                }

                cancelledKeys = 0;
                needsToSelectAgain = false;
                final int ioRatio = this.ioRatio;
                if (ioRatio == 100) {
                    try {
                        processSelectedKeys();
                    } finally {
                        runAllTasks();
                    }
                } else {
                    final long ioStartTime = System.nanoTime();
                    try {
                        processSelectedKeys();
                    } finally {
                        final long ioTime = System.nanoTime() - ioStartTime;
                        runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
            try {
                if (isShuttingDown()) {
                    closeAll();
                    if (confirmShutdown()) {
                        return;
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }

轮询 I/O 事件

selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())

// Netty 的任务队列包括普通任务、定时任务以及尾部任务，hasTask() 判断的是普通任务队列和尾部队列是否为空，而 delayNanos(currentTimeNanos) 方法获取的是定时任务的延迟时间。
hasTasks()

    @Override
    protected boolean hasTasks() {
        return super.hasTasks() || !tailTasks.isEmpty();
    }

// 选择策略返回值
// 如果包含普通任务和尾部任务，则调用selectNowSupplier的值；当 NioEventLoop 线程的不存在异步任务，即任务队列为空，返回的是 SELECT 策略
    @Override
    public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception {
        return hasTasks ? selectSupplier.get() : SelectStrategy.SELECT;
    }

// 如果当前 NioEventLoop 线程存在异步任务，会通过 selectSupplier.get() 最终调用到 selectNow() 方法，selectNow() 是非阻塞，执行后立即返回。
    private final IntSupplier selectNowSupplier = new IntSupplier() {
        @Override
        public int get() throws Exception {
            return selectNow();
        }
    };

    int selectNow() throws IOException {
        try {
            return selector.selectNow();
        } finally {
            // restore wakeup state if needed
            if (wakenUp.get()) {
                selector.wakeup();
            }
        }
    }


// 策略分两种情况
// 1、调用selectNow非阻塞查询，有就绪的channel，则直接跳到break执行下面的逻辑
// 2、调用selectNow无就绪结果，返回SELECT策略，调用select(wakenUp.getAndSet(false));

switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                    case SelectStrategy.CONTINUE:
                        continue;
                    case SelectStrategy.BUSY_WAIT:
                    case SelectStrategy.SELECT:
                        select(wakenUp.getAndSet(false));
                        if (wakenUp.get()) {
                            selector.wakeup();
                        }
                    default:
                }

调用select阻塞方法

select(boolean oldWakenUp)

// 1、计算 select 阻塞操作的最后截止时间，delayNanos为最近的一个定时任务，如果没有默认是一分钟
long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos); 

// 2、检测 select 阻塞操作是否超过截止时间，timeoutMillis<0说明有定时任务要执行，需要立即退出；但是预留了0.5ms的窗口时间，为了方便舍去小数
long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;

// 3、轮询过程中及时处理产生的任务
if (hasTasks() && wakenUp.compareAndSet(false, true)) {
       selector.selectNow();
       selectCnt = 1;//将已轮训次数重置为1
       break;
 }

// 4、将selector阻塞timeoutMillis毫秒
// 这里如果定时任务时间过长，肯定不能等待很长时间
// 在任务添加的时候，会调用wakeup方法唤醒线程，避免等待时间过长
int selectedKeys = selector.select(timeoutMillis);

// 5、满足一下任意条件，则推吹循环
/**
            * selectedKeys != 0 表示轮询到有已经就绪的IO事件
            * wakenUp.get() 表示是否被用户唤醒
            * hasTasks() 表示普通任务队列中有未完成的任务
            * hasScheduledTasks() 表示定时任务队列中有未完成的任务
            * 上述条件任何一个条件为真,则退出select，准备处理对应任务
            */
if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
                break;
            }

// 6、执行时间 >= 超时时间，说明selector执行正常，轮询次数置为1
if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
                selectCnt = 1;
            } 

// 7、执行时间 <超时时间，并且轮询次数达到阈值，说明发生空轮询，重新构建selector并关联channel
else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
                    selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
                logger.warn(
                        "Selector.select() returned prematurely {} times in a row; rebuilding Selector {}.",
                        selectCnt, selector);

                rebuildSelector();
                selector = this.selector;

                selector.selectNow();
                selectCnt = 1;
                break;
            }

处理IO事件

// 处理IO事件有两种方式，只看一下processSelectedKeysPlain
// 1、processSelectedKeysOptimized，处理Netty 优化过的 selectedKeys
// 2、processSelectedKeysPlain，正常的处理逻辑

// 1、处理连接事件。表示 TCP 连接建立成功, Channel 处于 Active 状态
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
            int ops = k.interestOps();
            ops &= ~SelectionKey.OP_CONNECT;
            k.interestOps(ops);
            unsafe.finishConnect();
        }
// 2、处理可写事件。表示上层可以向 Channel 写入数据，通过执行 ch.unsafe().forceFlush() 操作，将数据冲刷到客户端，最终会调用 javaChannel 的 write() 方法执行底层写操作
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
            ch.unsafe().forceFlush();
        }
// 3、处理可读事件。可读事件。表示 Channel 收到了可以被读取的新数据。依次调用 ChannelHandler 的 channelRead() 方法处理数据
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            unsafe.read();
        }

任务执行

// 1、将定时任务和普通任务合并
fetchedAll = fetchFromScheduledTaskQueue();
// 2、循环执行任务，就是直接调用的 Runnable 的 run() 方法。
safeExecute(task); // 执行任务
// 3、收尾工作，执行尾部队列任务，并不常用，一般用于统计
    protected void afterRunningAllTasks() {
        runAllTasksFrom(tailTasks);
    }

数据在 Pipeline 中的运转
1、数据读

// 1、首先通过EventLoop处理读事件，读取完成后触发channelRead
pipeline.fireChannelRead(byteBuf);

// 2、Pipeline的fireChannelRead调用了公用的invokeChannelRead，因为要从Head触发，所以此时传入的next为Head
@Override
    public final ChannelPipeline fireChannelRead(Object msg) {
        AbstractChannelHandlerContext.invokeChannelRead(head, msg);
        return this;
    }


// 3、调用channelContext的invokeChannelRead
    static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
        final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelRead(m);
        } else {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeChannelRead(m);
                }
            });
        }
    }

// 4、调用context绑定handler的channelRead，处理具体逻辑
    private void invokeChannelRead(Object msg) {
        if (invokeHandler()) {
            try {
                ((ChannelInboundHandler) handler()).channelRead(this, msg);
            } catch (Throwable t) {
                notifyHandlerException(t);
            }
        } else {
            fireChannelRead(msg);
        }
    }

// 5、如果没有实现channelRead方法，则选取下一个context，递归执行第三步
    @Override
    public ChannelHandlerContext fireChannelRead(final Object msg) {
        invokeChannelRead(findContextInbound(), msg);
        return this;
    }

2、数据写

// 1、调用writeAndFlush写数据，无论是从哪里开始写，都会调用Tail的writeAndFlush
    @Override
    public final ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {
        return tail.writeAndFlush(msg, promise);
    }

// 2、与读数据相反，写数据查找下一个节点是通过查找prev指针
    private AbstractChannelHandlerContext findContextOutbound() {
        AbstractChannelHandlerContext ctx = this;
        do {
            ctx = ctx.prev;
        } while (!ctx.outbound);
        return ctx;
    }

// 3、递归调用到HeadContext，通过unsafe写入socket
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            unsafe.write(msg, promise);
        }

3、异常处理

// 1 、发生异常后，调用fireExceptionCaught
pipeline.fireExceptionCaught(e);

// 2、从HeadContext开始传递异常信息
    @Override
    public final ChannelPipeline fireExceptionCaught(Throwable cause) {
        AbstractChannelHandlerContext.invokeExceptionCaught(head, cause);
        return this;
    }

// 3、同样的递归调用到TailContext的exceptionCaught，作为兜底逻辑，打印日志并释放资源
protected void onUnhandledInboundException(Throwable cause) {
        try {
            logger.warn(
                    "An exceptionCaught() event was fired, and it reached at the tail of the pipeline. " +
                            "It usually means the last handler in the pipeline did not handle the exception.",
                    cause);
        } finally {
            ReferenceCountUtil.release(cause);
        }
    }

// 4、由于异常信息是从HeadContext开始传递，所以最佳实践是将自定义的异常处理器放在TailContext前、其他自定义Handler后。可以保证如论是哪里发生了异常，自定义异常处理器都可以处理到

Netty处理粘包拆包

TCP 传输协议是面向流的，没有数据包界限。可能受MTU 传输单元大小、MSS 最大分段大小、滑动窗口等因素影响，将一个完整的报文拆分成多个小报文进行发送，也可能将多个报文合并成一个大的报文进行发送。因此就有了拆包和粘包。

ByteToMessageDecoder抽象类

// 1、ByteToMessageDecoder继承ChannelInboundHandlerAdapter，可以看出ByteToMessageDecoder解码器应该放在HeadContext后、其他Handler前的位置

// 2、当channel可读时，触发pipeline的channelRead实现，对应到ByteToMessageDecoder 的实现如下：
               public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//......省略其他内容
                  CodecOutputList out = CodecOutputList.newInstance();
            try {
                // channelRead的msg是ByteBuf类型
                ByteBuf data = (ByteBuf) msg;
                first = cumulation == null;
               // 如果是第一次，则初始化cumulation
                if (first) {
                    cumulation = data;
                } else {
                    // 不是第一次，追加到cumulation
                    cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data);
                }

                // 对数据进行处理
                callDecode(ctx, cumulation, out);

//......省略其他内容
    }

    protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        try {
            while(true) {
                if (in.isReadable()) {
                    // out中存放的是完整的对象集合，如果不为空，就可以接着触发channelRead
                    int outSize = out.size();
                    if (outSize > 0) {
                        // 循环推进out集合的内容
                        fireChannelRead(ctx, out, outSize);
                        out.clear();
                        if (ctx.isRemoved()) {
                            return;
                        }

                        outSize = 0;
                    }

                    int oldInputLength = in.readableBytes();
  
                    // 解码的核心逻辑
                    this.decodeRemovalReentryProtection(ctx, in, out);
                 //......省略
                }

                return;
            }
        } catch (DecoderException var6) {
            throw var6;
        } catch (Exception var7) {
            throw new DecoderException(var7);
        }
    }

final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, ByteBuf in, List<Object> out)
            throws Exception {
        // 设置当前状态为正在解码
        decodeState = STATE_CALLING_CHILD_DECODE;
        try {
            // 解码
            decode(ctx, in, out);
        } finally {
            // 执行hander的remove操作
            boolean removePending = decodeState == STATE_HANDLER_REMOVED_PENDING;
            decodeState = STATE_INIT;
            if (removePending) {
                handlerRemoved(ctx);
            }
        }
    }

    // 子类都重写了该方法，每种实现都会有自己特殊的解码方式
    // 有FixedLengthFrameDecoder、DelimiterBasedFrameDecoder、LengthFieldBasedFrameDecoder
    protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;

最简单的解决方案：FixedLengthFrameDecoder

@Override
    protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        Object decoded = decode(ctx, in);
        if (decoded != null) {
           // 如果有符合要求的对象，添加到集合中，由父类继续推动channelRead，此时的对象是一个完整的对象
            out.add(decoded);
        }
    }

    protected Object decode(
            @SuppressWarnings("UnusedParameters") ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        // 收集到的数据是否小于固定长度，小于就代表无法解析，直接返回空；并等待ByteBuf中的数据，使能够达到固定长度
        
        if (in.readableBytes() < frameLength) {
            return null;
        } else {
            return in.readRetainedSlice(frameLength);
        }
    }

Netty学习整理

1 复习NIO

2 Netty整体架构

2.1 网络通信层

2.2 事件调度层

2.3 服务编排层

3 服务启动流程

4 处理请求流程

Netty处理粘包拆包

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读