Netty 基础入门

文章目录

• • 一 基本介绍
  • • 1.1 Netty知识了解
    • • 1.1.1 概述
      • 1.1.2 设计方面
      • 1.1.3 易用性
      • 1.1.4 性能
      • 1.1.5 安全性
    • 1.2 为啥学习Netty
    • • 1.2.1 传统的HTTP
      • 1.2.2 并发线程下的NIO
      • 1.2.3 Netty
    • 1.3 Reactor线程模型
    • 1.4 基本执行流程
  • 二 组件基本使用
  • • 2.1 EventLoop
    • • 2.1.1 处理普通任务与定时任务
      • 2.1.2 处理IO任务
    • 2.2 Channel
    • • 2.2.1 ChannelFuture
      • 2.2.2 CloseFuture
    • 2.3 Future&Promise
    • • 2.3.1 Jdk Future
      • 2.3.2 Netty Future
      • 2.3.3 Netty Promise
    • 2.4 Handler&Pipeline
    • • 2.4.1 基本知识了解
    • 2.5 ByteBuf
    • • 2.5.1 基本创建
      • 2.5.2 直接内存与堆内存
      • 2.5.3 ByteBuf基本结构
      • 2.5.4 组成
      • 2.5.5 写方法
      • 2.5.6 扩容
      • 2.5.7 读取方法
      • 2.5.8 内存释放
      • 2.5.9 零拷贝
  • 三 进阶部分
  • • 3.1 半包与粘包问题
    • • 3.1.1 现象
      • 3.1.2 原因
      • 3.1.3 解决方法
      • • 3.1.3.1 短连接
        • 3.1.3.2 定长解码器
        • 3.1.3.3 行解码器
        • 3.1.3.4 长度字段解码器
    • 3.2 协议的设计
    • • 3.2.1 Redis协议
      • 3.2.2 Http协议
      • 3.2.3 自定义协议
      • • 3.2.3.1 要素
        • 3.2.3.2 代码
    • 3.3 重要参数
    • • CONNECT_TIMEOUT_MILLIS
      • SO_BACKLOG
      • TCP_NODELAY
      • SO_SNDBUF & SO_RCVBUF
      • ALLOCATOR
      • RCVBUF_ALLOCATOR

一 基本介绍

1.1 Netty知识了解

1.1.1 概述

官网：https://netty.io/

• Netty 是一个 NIO 客户端服务器框架，它可以快速轻松地开发网络应用程序，例如协议服务器和客户端。它极大地简化和精简了 TCP 和 UDP 套接字服务器等网络编程。
• Netty 是基于 Java NIO 的异步事件驱动的网络应用框架，使用 Netty 可以快速开发网络应用，Netty 提供了高层次的抽象来简化 TCP 和 UDP 服务器的编程，但是你仍然可以使用底层的 API。
• Netty 的内部实现是很复杂的，但是 Netty 提供了简单易用的API从网络处理代码中解耦业务逻辑。Netty 是完全基于 NIO 实现的，所以整个 Netty 都是异步的。
• Netty 是最流行的 NIO 框架，它已经得到成百上千的商业、商用项目验证，许多框架和开源组件的底层 rpc 都是使用的 Netty，如 Dubbo、Elasticsearch 等等。下面是官网给出的一些 Netty 的特性：

1.1.2 设计方面

• 对各种传输协议提供统一的 API（使用阻塞和非阻塞套接字时候使用的是同一个 API，只是需要设置的参数不一样）。
• 基于一个灵活、可扩展的事件模型来实现关注点清晰分离。
• 高度可定制的线程模型——单线程、一个或多个线程池。
• 真正的无数据报套接字（UDP）的支持（since 3.1）。

1.1.3 易用性

• 完善的 Javadoc 文档和示例代码。
• 不需要额外的依赖，JDK 5 (Netty 3.x) 或者 JDK 6 (Netty 4.x) 已经足够。

1.1.4 性能

• 更好的吞吐量，更低的等待延迟。
• 更少的资源消耗。
• 最小化不必要的内存拷贝。

1.1.5 安全性

• 完整的 SSL/TLS 和 StartTLS 支持

1.2 为啥学习Netty

1.2.1 传统的HTTP

• 创建一个ServerSocket，监听并绑定一个端口
• 一系列客户端来请求这个端口
• 服务器使用Accept，获得一个来自客户端的Socket连接对象
• 启动一个新线程处理连接
  • 读Socket，得到字节流
  • 解码协议，得到Http请求对象
  • 处理Http请求，得到一个结果，封装成一个HttpResponse对象
  • 编码协议，将结果序列化字节流
  • 写Socket，将字节流发给客户端
  • 继续循环处理http请求
• HTTP服务器之所以称为HTTP服务器，是因为编码解码协议是HTTP协议，如果协议是Redis协议，那它就成了Redis服务器，如果协议是WebSocket，那它就成了WebSocket服务器，等等。
• 使用Netty你就可以定制编解码协议，实现自己的特定协议的服务器。

1.2.2 并发线程下的NIO

• NIO的全称是NoneBlocking IO，非阻塞IO，区别与BIO，BIO的全称是Blocking IO，阻塞IO。那这个阻塞是什么意思呢？
  1. Accept是阻塞的，只有新连接来了，Accept才会返回，主线程才能继
  2. Read是阻塞的，只有请求消息来了，Read才能返回，子线程才能继续处理
  3. Write是阻塞的，只有客户端把消息收了，Write才能返回，子线程才能继续读取下一个请求
• 所以传统的多线程服务器是BlockingIO模式的，从头到尾所有的线程都是阻塞的。这些线程就干等在哪里，占用了操作系统的调度资源，什么事也不干，是浪费。
• 那么NIO是怎么做到非阻塞的呢。它用的是事件机制。它可以用一个线程把Accept，读写操作，请求处理的逻辑全干了。如果什么事都没得做，它也不会死循环，它会将线程休眠起来，直到下一个事件来了再继续干活，这样的一个线程称之为NIO线程。
• 如果了解NIO的详细知识，请参考前面的博客

1.2.3 Netty

• Netty是建立在NIO基础之上，Netty在NIO之上又提供了更高层次的抽象。
• 在Netty里面，Accept连接可以使用单独的线程池去处理，读写操作又是另外的线程池来处理。
• Accept连接和读写操作也可以使用同一个线程池来进行处理。而请求处理逻辑既可以使用单独的线程池进行处理，也可以跟放在读写线程一块处理。线程池中的每一个线程都是NIO线程。用户可以根据实际情况进行组装，构造出满足系统需求的并发模型。

1.3 Reactor线程模型

Reactor单线程模型

一个NIO线程+一个accept线程：

Reactor多线程模型

Reactor主从模型

主从Reactor多线程：多个acceptor的NIO线程池用于接受客户端的连接

1.4 基本执行流程

• 服务端

package com.shu.NettyHello;

import com.sun.scenario.effect.impl.sw.sse.SSEBlend_SRC_OUTPeer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;

/**
 * @Author shu
 * @Date: 2022/02/26/ 13:36
 * @Description
 **/
public class HelloServer {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new StringDecoder());
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                super.channelRead(ctx, msg);
                                System.out.println(msg);
                            }
                        });
                    }
                })
                .bind(8082);
    }
}

• 客服端

package com.shu.NettyHello;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;

/**
 * @Author shu
 * @Date: 2022/02/26/ 13:36
 * @Description
 **/
public class HelloClient {
    public static void main(String[] args) throws InterruptedException {
        new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new StringEncoder());

                    }
                }).bind(new InetSocketAddress("localhost",8082))
                .sync()
                .channel()
                .writeAndFlush("hello word");
    }
}

二 组件基本使用

2.1 EventLoop

• 事件循环对象 EventLoop 本质是一个单线程执行器（同时维护了一个 Selector），里面有 run 方法处理一个或多个 Channel 上源源不断的 io 事件
• 注册后将处理Channel所有 I/O 操作。 一个EventLoop实例通常会处理多个Channel但这可能取决于实现细节和内部结构。
• 继承自 j.u.c.ScheduledExecutorService 因此包含了线程池中所有的方法
• 继承自 netty 自己的 OrderedEventExecutor
  • 提供了 boolean inEventLoop(Thread thread) 方法判断一个线程是否属于此 EventLoop
  • 提供了 EventLoopGroup parent() 方法来看看自己属于哪个 EventLoopGroup

• 事件循环组 EventLoopGroup 是一组 EventLoop，Channel 一般会调用 EventLoopGroup 的 register 方法来绑定其中一个 EventLoop，后续这个 Channel 上的 io 事件都由此 EventLoop 来处理（保证了 io 事件处理时的线程安全）
• 继承自 netty 自己的 EventExecutorGroup
  • 实现了 Iterable 接口提供遍历 EventLoop 的能力
  • 另有 next 方法获取集合中下一个 EventLoop

2.1.1 处理普通任务与定时任务

package com.shu.EventLoop;


import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.TimeUnit;


/**
 * @Author shu
 * @Date: 2022/03/01/ 16:20
 * @Description
 **/
@Slf4j
public class NioEventLoopTest {
    public static void main(String[] args) {
        EventLoopGroup eventLoopGroup=new NioEventLoopGroup(2);
        // 执行普通任务
        eventLoopGroup.next().submit(()->{

        }
        );
        // 执行定时任务
        eventLoopGroup.next().scheduleAtFixedRate(()->{
            System.out.println("ok");
        },0,1, TimeUnit.SECONDS);
        
         // 优雅地关闭
        eventLoopGroup.shutdownGracefully();
    }
}

2.1.2 处理IO任务

• 服务端

package com.shu.EventLoop;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import java.nio.charset.StandardCharsets;

/**
 * @Author shu
 * @Date: 2022/03/01/ 16:37
 * @Description 自定义EventLoopGroup
 **/
public class NioEventLoopServer {
    public static void main(String[] args) {
        // 处理读，写事件
        EventLoopGroup defaultEventLoop=new DefaultEventLoop();
        new ServerBootstrap()
            //  两个Group，分别为Boss 负责Accept事件，Worker 负责读写事件
                .group(new NioEventLoopGroup(1),new NioEventLoopGroup(2))
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        // 增加两个handler，第一个使用NioEventLoopGroup处理，第二个使用自定义EventLoopGroup处理
                        socketChannel.pipeline().addLast("nioHandler",new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf buf = (ByteBuf) msg;
                                System.out.println(Thread.currentThread().getName() + " " + buf.toString(StandardCharsets.UTF_8));
                                // 调用下一个handler
                                ctx.fireChannelRead(msg);
                            }
                            })
                                // 该handler绑定自定义的Group
                                .addLast(defaultEventLoop, "myHandler", new ChannelInboundHandlerAdapter() {
                                    @Override
                                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                        ByteBuf buf = (ByteBuf) msg;
                                        System.out.println(Thread.currentThread().getName() + " " + buf.toString(StandardCharsets.UTF_8));
                                    }
                                });
                    }
                })
                .bind(8080);
    }
}

• 客服端

package com.shu.EventLoop;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import java.io.IOException;
import java.net.InetSocketAddress;

/**
 * @Author shu
 * @Date: 2022/03/01/ 16:37
 * @Description 客服端
 **/
public class NioEventLoopClient {
    public static void main(String[] args) throws IOException, InterruptedException {
        Channel channel = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new StringEncoder());
                    }
                })
                .connect(new InetSocketAddress("localhost", 8080))
                .sync()
                .channel();
        System.out.println(channel);
        // 此处打断点调试，调用 channel.writeAndFlush(...);
        System.in.read();
    }
}

• 可以看出，一个EventLoop可以负责多个Channel，且EventLoop一旦与Channel绑定，则一直负责处理该Channel中的事件
• 当有的任务需要较长的时间处理时，可以使用非NioEventLoopGroup，避免同一个NioEventLoop中的其他Channel在较长的时间内都无法得到处理

2.2 Channel

常用方法

• close() 可以用来关闭Channel
• closeFuture() 用来处理 Channel 的关闭
  • sync 方法作用是同步等待 Channel 关闭
  • 而 addListener 方法是异步等待 Channel 关闭
• pipeline() 方法用于添加处理器
• write() 方法将数据写入
  • 因为缓冲机制，数据被写入到 Channel 中以后，不会立即被发送
  • 只有当缓冲满了或者调用了flush()方法后，才会将数据通过 Channel 发送出去
• writeAndFlush() 方法将数据写入并立即发送（刷出）
• id():返回此通道的全局唯一标识符。
• isActive():如果通道处于活动状态并连接,则返回true.。
• isOpen():如果通道打开并且可能稍后激活,则返回true。
• isRegistered():如果通道注册了EventLoop，则返回true。
• config():返回关于此通道的配置。
• localAddress():返回此通道绑定的本地地址。
• pipeline():返回分派的ChannelPipeline。
• remoteAddress():返回此通道连接到的远程地址。

2.2.1 ChannelFuture

同步发送：sync()

public class ChannelClient {
    public static void main(String[] args) throws InterruptedException {
        ChannelFuture channelFuture = new Bootstrap()
                .group(new NioEventLoopGroup())
                // 选择客户 Socket 实现类，NioSocketChannel 表示基于 NIO 的客户端实现
                .channel(NioSocketChannel.class)
                // ChannelInitializer 处理器（仅执行一次）
                // 它的作用是待客户端SocketChannel建立连接后，执行initChannel以便添加更多的处理器
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        // 消息会经过通道 handler 处理，这里是将 String => ByteBuf 编码发出
                        channel.pipeline().addLast(new StringEncoder());
                    }
                })
                // 指定要连接的服务器和端口
                .connect(new InetSocketAddress("localhost", 8080));

//        // 同步处理
        ChannelFuture future = channelFuture.sync();
        future.channel().writeAndFlush("hello word");
    }
}

• 如果我们去掉channelFuture.sync()方法，会服务器无法收到hello world。
• 这是因为建立连接(connect)的过程是异步非阻塞的，若不通过sync()方法阻塞主线程，等待连接真正建立，这时通过 channelFuture.channel() 拿到的 Channel 对象，并不是真正与服务器建立好连接的 Channel，也就没法将信息正确的传输给服务器端。
• 所以需要通过channelFuture.sync()方法，阻塞主线程，同步处理结果，等待连接真正建立好以后，再去获得 Channel 传递数据。
• 使用该方法，获取 Channel 和发送数据的线程都是主线程。

异步发送：addListener方法
​

package com.shu.Channel;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;

import java.net.InetSocketAddress;

/**
 * @Author shu
 * @Date: 2022/03/02/ 11:29
 * @Description
 **/
public class ChannelClient {
    public static void main(String[] args) throws InterruptedException {
        ChannelFuture channelFuture = new Bootstrap()
                .group(new NioEventLoopGroup())
                // 选择客户 Socket 实现类，NioSocketChannel 表示基于 NIO 的客户端实现
                .channel(NioSocketChannel.class)
                // ChannelInitializer 处理器（仅执行一次）
                // 它的作用是待客户端SocketChannel建立连接后，执行initChannel以便添加更多的处理器
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        // 消息会经过通道 handler 处理，这里是将 String => ByteBuf 编码发出
                        channel.pipeline().addLast(new StringEncoder());
                    }
                })
                // 指定要连接的服务器和端口
                .connect(new InetSocketAddress("localhost", 8080));


        // 异步处理
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                future.channel().writeAndFlush("hello word");
            }
        });


    }
}

• 通过这种方法可以在NIO线程中获取 Channel 并发送数据，而不是在主线程中执行这些操作​

2.2.2 CloseFuture

当我们要关闭channel时，可以调用channel.close()方法进行关闭。但是该方法也是一个异步方法。真正的关闭操作并不是在调用该方法的线程中执行的，而是在NIO线程中执行真正的关闭操作
同步关闭：closeFuture.sync();

package com.shu.Channel;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

import java.net.InetSocketAddress;
import java.util.Scanner;

/**
 * @Author shu
 * @Date: 2022/03/02/ 11:29
 * @Description
 **/
public class ChannelClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();
        ChannelFuture channelFuture = new Bootstrap()
                .group(group)
                // 选择客户 Socket 实现类，NioSocketChannel 表示基于 NIO 的客户端实现
                .channel(NioSocketChannel.class)
                // ChannelInitializer 处理器（仅执行一次）
                // 它的作用是待客户端SocketChannel建立连接后，执行initChannel以便添加更多的处理器
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        // 消息会经过通道 handler 处理，这里是将 String => ByteBuf 编码发出
                        channel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        channel.pipeline().addLast(new StringEncoder());
                    }
                })
                // 指定要连接的服务器和端口
                .connect(new InetSocketAddress("localhost", 8080));

        Channel channel = channelFuture.channel();
        Scanner scanner = new Scanner(System.in);

        // 创建一个线程用于输入并向服务器发送
        new Thread(()->{
            while (true) {
                String msg = scanner.next();
                if ("q".equals(msg)) {
                    // 关闭操作是异步的，在NIO线程中执行
                    channel.close();
                    break;
                }
                channel.writeAndFlush(msg);
            }
        }, "inputThread").start();


        // 获得closeFuture对象
        ChannelFuture closeFuture = channel.closeFuture();
        System.out.println("waiting close...");
        // 同步等待NIO线程执行完close操作
        closeFuture.sync();
        // 关闭之后执行一些操作，可以保证执行的操作一定是在channel关闭以后执行的
        System.out.println("关闭之后执行一些额外操作...");
        // 关闭EventLoopGroup
        group.shutdownGracefully();
    }
}

异步关闭：addListener方法

package com.shu.Channel;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;

import java.net.InetSocketAddress;
import java.util.Scanner;

/**
 * @Author shu
 * @Date: 2022/03/02/ 11:29
 * @Description
 **/
public class ChannelClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();
        ChannelFuture channelFuture = new Bootstrap()
                .group(group)
                // 选择客户 Socket 实现类，NioSocketChannel 表示基于 NIO 的客户端实现
                .channel(NioSocketChannel.class)
                // ChannelInitializer 处理器（仅执行一次）
                // 它的作用是待客户端SocketChannel建立连接后，执行initChannel以便添加更多的处理器
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        // 消息会经过通道 handler 处理，这里是将 String => ByteBuf 编码发出
                        channel.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        channel.pipeline().addLast(new StringEncoder());
                    }
                })
                // 指定要连接的服务器和端口
                .connect(new InetSocketAddress("localhost", 8080));

        Channel channel = channelFuture.channel();
        Scanner scanner = new Scanner(System.in);

        // 创建一个线程用于输入并向服务器发送
        new Thread(()->{
            while (true) {
                String msg = scanner.next();
                if ("q".equals(msg)) {
                    // 关闭操作是异步的，在NIO线程中执行
                    channel.close();
                    break;
                }
                channel.writeAndFlush(msg);
            }
        }, "inputThread").start();


        // 获得closeFuture对象
        ChannelFuture closeFuture = channel.closeFuture();
        System.out.println("waiting close...");

        // 异步关闭
        closeFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                // 等待channel关闭后才执行的操作
                System.out.println("关闭之后执行一些额外操作...");
                // 关闭EventLoopGroup
                group.shutdownGracefully();
            }
        });


    }
}

2.3 Future&Promise

2.3.1 Jdk Future

• 基本方法

public interface Future<V> {
	// 取消任务
    boolean cancel(boolean mayInterruptIfRunning);
	// 是否取消
    boolean isCancelled();
	// 是否完成
    boolean isDone();
	// 获取结果
    V get() throws InterruptedException, ExecutionException;
	// 超时获取结果
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

• 多线程中利用Future返回值

package com.shu.Channel;

import java.util.concurrent.*;

/**
 * @Author shu
 * @Date: 2022/03/02/ 14:01
 * @Description 多线程中返回任务
 **/
public class JdkFutureTask {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(2);

        Future<Integer> future = executorService.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return 2;
            }
        });

        Integer integer = future.get();

        System.out.println(integer);
    }
}

2.3.2 Netty Future

• 基本方法

public interface Future<V> extends java.util.concurrent.Future<V> {

  	// 当且仅当 I/O 操作成功完成时返回true 。
    boolean isSuccess();

    // 当且仅当可以通过cancel(boolean)操作时才返回true 。
    boolean isCancellable();

    // 如果 I/O 操作失败，则返回 I/O 操作失败的原因。
    Throwable cause();

    // 将指定的侦听器添加到这个未来。当这个未来完成时，会通知指定的侦听器。如果这个 future 已经完成，则立即通知指定的侦听器。
    Future<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);

    Future<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners);

    // 从此未来中删除指定侦听器的第一次出现。当这个未来完成时，不再通知指定的侦听器。如果指定的侦听器与此未来无关，则此方法不执行任何操作并静默返回。
    Future<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);

    
    Future<V> removeListeners(GenericFutureListener<? extends Future<? super V>>... listeners);

    // 等待这个未来，直到它完成，如果这个未来失败，则重新抛出失败的原因。
    Future<V> sync() throws InterruptedException;

   	// 
    Future<V> syncUninterruptibly();

   // 等待这个未来完成。
    Future<V> await() throws InterruptedException;

    Future<V> awaitUninterruptibly();

   
    boolean await(long timeout, TimeUnit unit) throws InterruptedException;

   
    boolean await(long timeoutMillis) throws InterruptedException;

   
    boolean awaitUninterruptibly(long timeout, TimeUnit unit);

    
    boolean awaitUninterruptibly(long timeoutMillis);

  	// 不阻塞地返回结果。如果未来还没有完成，这将返回null 。由于可能使用null值将未来标记为成功，因此您还需要检查未来是否真的用isDone()完成，而不是依赖于返回的null值。
    V getNow();

    @Override
    boolean cancel(boolean mayInterruptIfRunning);
}

package com.shu.Channel;

import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;

/**
 * @Author shu
 * @Date: 2022/03/02/ 14:22
 * @Description Netty中的Future
 **/
public class NettyFutureTask {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();

        // 获得 EventLoop 对象
        EventLoop eventLoop = group.next();
        Future<Integer> future = eventLoop.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return 50;
            }
        });

        // 主线程中获取结果
        System.out.println(Thread.currentThread().getName() + " 获取结果");
        System.out.println("getNow " + future.getNow());
        System.out.println("get " + future.get());

        // NIO线程中异步获取结果
        future.addListener(new GenericFutureListener<Future<? super Integer>>() {
            @Override
            public void operationComplete(Future<? super Integer> future) throws Exception {
                System.out.println(Thread.currentThread().getName() + " 获取结果");
                System.out.println("getNow " + future.getNow());
            }
        });
    }
}

Netty中的Future对象，可以通过EventLoop的sumbit()方法得到

• 可以通过Future对象的get方法，阻塞地获取返回结果
• 也可以通过getNow方法，获取结果，若还没有结果，则返回null，该方法是非阻塞的
• 还可以通过future.addListener方法，在Callable方法执行的线程中，异步获取返回结果

2.3.3 Netty Promise

• 基本方法

public interface Promise<V> extends Future<V> {

   
    Promise<V> setSuccess(V result);

   
    boolean trySuccess(V result);

    Promise<V> setFailure(Throwable cause);

   
    boolean tryFailure(Throwable cause);

    
    boolean setUncancellable();

    @Override
    Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);

    @Override
    Promise<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners);

    @Override
    Promise<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);

    @Override
    Promise<V> removeListeners(GenericFutureListener<? extends Future<? super V>>... listeners);

    @Override
    Promise<V> await() throws InterruptedException;

    @Override
    Promise<V> awaitUninterruptibly();

    @Override
    Promise<V> sync() throws InterruptedException;

    @Override
    Promise<V> syncUninterruptibly();
}

2.4 Handler&Pipeline

2.4.1 基本知识了解

• ChannelHandler用来处理Channel 上的各种事件，分为入站、出站两种。所有ChannelHandler被连成—串，就是 Pipeline
• ·入站处理器通常是ChannellnboundHandlerAdapter的子类，主要用来读取客户端数据，写回结果。
• 出站处理器通常是ChannelOutboundHandlerAdapter 的子类，主要对写回结果进行加工。
• 打个比喻，每个Channel是一个产品的加工车间,Pipeline是车间中的流水线，ChannelHandler就是流水线上的各道工序，而后面要讲的ByteBuf是原材料，经过很多工序的加工:先经过一道道入站工序，再经过一道道出站工序最终变成产品。

public class PipeLineServer {
    public static void main(String[] args) {
        new ServerBootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        // 在socketChannel的pipeline中添加handler
                        // pipeline中handler是带有head与tail节点的双向链表，的实际结构为
    				 	// head <-> handler1 <-> ... <-> handler4 <->tail
                        // Inbound主要处理入站操作，一般为读操作，发生入站操作时会触发Inbound方法
                        // 入站时，handler是从head向后调用的
                        socketChannel.pipeline().addLast("handler1" ,new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                System.out.println(Thread.currentThread().getName() + " Inbound handler 1");
                                // 父类该方法内部会调用fireChannelRead
                                // 将数据传递给下一个handler
                                super.channelRead(ctx, msg);
                            }
                        });
                        socketChannel.pipeline().addLast("handler2", new ChannelInboundHandlerAdapter() {
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                System.out.println(Thread.currentThread().getName() + " Inbound handler 2");
                                // 执行write操作，使得Outbound的方法能够得到调用
          socketChannel.writeAndFlush(ctx.alloc().buffer().writeBytes("Server...".getBytes(StandardCharsets.UTF_8)));
                                super.channelRead(ctx, msg);
                            }
                        });
                        // Outbound主要处理出站操作，一般为写操作，发生出站操作时会触发Outbound方法
                        // 出站时，handler的调用是从tail向前调用的
                        socketChannel.pipeline().addLast("handler3" ,new ChannelOutboundHandlerAdapter(){
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                System.out.println(Thread.currentThread().getName() + " Outbound handler 1");
                                super.write(ctx, msg, promise);
                            }
                        });
                        socketChannel.pipeline().addLast("handler4" ,new ChannelOutboundHandlerAdapter(){
                            @Override
                            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                                System.out.println(Thread.currentThread().getName() + " Outbound handler 2");
                                super.write(ctx, msg, promise);
                            }
                        });
                    }
                })
                .bind(8080);
    }
}

• pipeline是结构是一个带有head与tail指针的双向链表，其中的节点为handler
  • 要通过ctx.fireChannelRead(msg)等方法，将当前handler的处理结果传递给下一个handler
• 当有入站（Inbound）操作时，会从head开始向后调用handler，直到handler不是处理Inbound操作为止
• 当有出站（Outbound）操作时，会从tail开始向前调用handler，直到handler不是处理Outbound操作为止

2.5 ByteBuf

2.5.1 基本创建

• ByteBuf通过ByteBufAllocator选择allocator并调用对应的buffer()方法来创建的，默认使用直接内存作为ByteBuf，容量为256个字节，可以指定初始容量的大小
• 当ByteBuf的容量无法容纳所有数据时，ByteBuf会进行扩容操作，如果在handler中创建ByteBuf，建议使用ChannelHandlerContext ctx.alloc().buffer()来创建

public class ByteBufStudy {
    public static void main(String[] args) {
        // 创建ByteBuf
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16);
        ByteBufUtil.log(buffer);

        // 向buffer中写入数据
        StringBuilder sb = new StringBuilder();
        for(int i = 0; i < 20; i++) {
            sb.append("a");
        }
        buffer.writeBytes(sb.toString().getBytes(StandardCharsets.UTF_8));

        // 查看写入结果
        ByteBufUtil.log(buffer);
    }

2.5.2 直接内存与堆内存

package com.shu.NettyByteBuf;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;

import java.nio.ByteBuffer;

/**
 * @Author shu
 * @Date: 2022/03/02/ 15:53
 * @Description
 **/
public class CreatByteBuf {
    public static void main(String[] args) {
        // 堆内存分配
        ByteBuf heapBuffer = ByteBufAllocator.DEFAULT.heapBuffer();
        // 直接内存分配
        ByteBuf directBuffer = ByteBufAllocator.DEFAULT.directBuffer();
    }
}

• 直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用
• 直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放

2.5.3 ByteBuf基本结构

• ByteBuf提供了两个指针变量来支持顺序读取和写入操作——分别用于读取操作的readerIndex和用于写入操作的writerIndex 。下图显示了缓冲区如何通过两个指针分割成三个区域：
  ±------------------±-----------------±-----------------+
  | discardable bytes | readable bytes | writable bytes |
  | | (CONTENT) | |
  ±------------------±-----------------±-----------------+
  | | | |
  0 <= readerIndex <= writerIndex <= capacity

• 可读字节（实际内容），该段是存储实际数据的地方。任何名称以read或skip开头的操作都将获取或跳过当前readerIndex处的数据，并将其增加读取字节数。如果读操作的参数也是一个ByteBuf并且没有指定目标索引，则指定缓冲区的writerIndex一起增加。

• 可写字节，该段是需要填充的未定义空间。任何名称以write开头的操作都将在当前writerIndex处写入数据，并将其增加写入的字节数。如果写操作的参数也是一个ByteBuf ，并且没有指定源索引，则指定缓冲区的readerIndex一起增加，如果没有足够的可写字节，则会引发IndexOutOfBoundsException 。新分配缓冲区的writerIndex的默认值为0 。包装或复制缓冲区的writerIndex的默认值是缓冲区的capacity 。

2.5.4 组成

ByteBuf主要有以下几个组成部分

• 最大容量与当前容量
  • 在构造ByteBuf时，可传入两个参数，分别代表初始容量和最大容量，若未传入第二个参数（最大容量），最大容量默认为Integer.MAX_VALUE
  • 当ByteBuf容量无法容纳所有数据时，会进行扩容操作，若超出最大容量，会抛出java.lang.IndexOutOfBoundsException异常
• 读写操作不同于ByteBuffer只用position进行控制，ByteBuf分别由读指针和写指针两个指针控制。进行读写操作时，无需进行模式的切换
  • 读指针前的部分被称为废弃部分，是已经读过的内容
  • 读指针与写指针之间的空间称为可读部分
  • 写指针与当前容量之间的空间称为可写部分

2.5.5 写方法

常用方法如下

注意

• 这些方法的未指明返回值的，其返回值都是 ByteBuf，意味着可以链式调用来写入不同的数据
• 网络传输中，默认习惯是 Big Endian，使用 writeInt(int value)

public class ByteBufStudy {
    public static void main(String[] args) {
        // 创建ByteBuf
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16, 20);
        ByteBufUtil.log(buffer);

        // 向buffer中写入数据
        buffer.writeBytes(new byte[]{1, 2, 3, 4});
        ByteBufUtil.log(buffer);

        buffer.writeInt(5);
        ByteBufUtil.log(buffer);

        buffer.writeIntLE(6);
        ByteBufUtil.log(buffer);

        buffer.writeLong(7);
        ByteBufUtil.log(buffer);
    }
}

2.5.6 扩容

• 当ByteBuf中的容量无法容纳写入的数据时，会进行扩容操作
• 如何写入后数据大小未超过 512 字节，则选择下一个 16 的整数倍进行扩容
  • 例如写入后大小为 12 字节，则扩容后 capacity 是 16 字节
• 如果写入后数据大小超过 512 字节，则选择下一个 2n
  • 例如写入后大小为 513 字节，则扩容后 capacity 是 210=1024 字节（29=512 已经不够了）
• 扩容不能超过 maxCapacity，否则会抛出java.lang.IndexOutOfBoundsException异常

2.5.7 读取方法

• 读取主要是通过一系列read方法进行读取，读取时会根据读取数据的字节数移动读指针
• 如果需要重复读取，需要调用buffer.markReaderIndex()对读指针进行标记，并通过buffer.resetReaderIndex()将读指针恢复到mark标记的位置

public class ByteBufStudy {
    public static void main(String[] args) {
        // 创建ByteBuf
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16, 20);

        // 向buffer中写入数据
        buffer.writeBytes(new byte[]{1, 2, 3, 4});
        buffer.writeInt(5);

        // 读取4个字节
        System.out.println(buffer.readByte());
        System.out.println(buffer.readByte());
        System.out.println(buffer.readByte());
        System.out.println(buffer.readByte());
        ByteBufUtil.log(buffer);

        // 通过mark与reset实现重复读取
        buffer.markReaderIndex();
        System.out.println(buffer.readInt());
        ByteBufUtil.log(buffer);

        // 恢复到mark标记处
        buffer.resetReaderIndex();
        ByteBufUtil.log(buffer);
    }
}

2.5.8 内存释放

由于 Netty 中有堆外内存（直接内存）的 ByteBuf 实现，堆外内存最好是手动来释放，而不是等 GC 垃圾回收。

• UnpooledHeapByteBuf 使用的是 JVM 内存，只需等 GC 回收内存即可
• UnpooledDirectByteBuf 使用的就是直接内存了，需要特殊的方法来回收内存
• PooledByteBuf 和它的子类使用了池化机制，需要更复杂的规则来回收内存

Netty 这里采用了引用计数法来控制回收内存，每个 ByteBuf 都实现了 ReferenceCounted 接口

• 每个 ByteBuf 对象的初始计数为 1
• 调用 release 方法计数减 1，如果计数为 0，ByteBuf 内存被回收

释放规则
因为 pipeline 的存在，一般需要将 ByteBuf 传递给下一个 ChannelHandler，如果在每个 ChannelHandler 中都去调用 release ，就失去了传递性（如果在这个 ChannelHandler 内这个 ByteBuf 已完成了它的使命，那么便无须再传递）
基本规则是，谁是最后使用者，谁负责 release

• 起点，对于 NIO 实现来讲，在 io.netty.channel.nio.AbstractNioByteChannel.NioByteUnsafe.read 方法中首次创建 ByteBuf 放入 pipeline（line 163 pipeline.fireChannelRead(byteBuf)）
• 入站 ByteBuf 处理原则
  • 对原始 ByteBuf 不做处理，调用 ctx.fireChannelRead(msg) 向后传递，这时无须 release
  • 将原始 ByteBuf 转换为其它类型的 Java 对象，这时 ByteBuf 就没用了，必须 release
  • 如果不调用 ctx.fireChannelRead(msg) 向后传递，那么也必须 release
  • 注意各种异常，如果 ByteBuf 没有成功传递到下一个 ChannelHandler，必须 release
  • 假设消息一直向后传，那么 TailContext 会负责释放未处理消息（原始的 ByteBuf）
• 出站 ByteBuf 处理原则
  • 出站消息最终都会转为 ByteBuf 输出，一直向前传，由 HeadContext flush 后 release
• 异常处理原则
  • 有时候不清楚 ByteBuf 被引用了多少次，但又必须彻底释放，可以循环调用 release 直到返回 truewhile (!buffer.release()) {}Copy

当ByteBuf被传到了pipeline的head与tail时，ByteBuf会被其中的方法彻底释放，但前提是ByteBuf被传递到了head与tail中

protected void onUnhandledInboundMessage(Object msg) {
    try {
        logger.debug("Discarded inbound message {} that reached at the tail of the pipeline. Please check your pipeline configuration.", msg);
    } finally {
        // 具体的释放方法
        ReferenceCountUtil.release(msg);
    }
}

2.5.9 零拷贝

• ByteBuf切片是【零拷贝】的体现之一，对原始 ByteBuf 进行切片成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制，还是使用原始 ByteBuf 的内存，切片后的 ByteBuf 维护独立的 read，write 指针
• 得到分片后的buffer后，要调用其retain方法，使其内部的引用计数加一。避免原ByteBuf释放，导致切片buffer无法使用，修改原ByteBuf中的值，也会影响切片后得到的ByteBuf

public class TestSlice {
    public static void main(String[] args) {
        // 创建ByteBuf
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer(16, 20);

        // 向buffer中写入数据
        buffer.writeBytes(new byte[]{1, 2, 3, 4, 5, 6, 7, 8, 9, 10});

        // 将buffer分成两部分
        ByteBuf slice1 = buffer.slice(0, 5);
        ByteBuf slice2 = buffer.slice(5, 5);

        // 需要让分片的buffer引用计数加一
        // 避免原Buffer释放导致分片buffer无法使用
        slice1.retain();
        slice2.retain();
        
        ByteBufUtil.log(slice1);
        ByteBufUtil.log(slice2);

        // 更改原始buffer中的值
        System.out.println("===========修改原buffer中的值===========");
        buffer.setByte(0,5);

        System.out.println("===========打印slice1===========");
        ByteBufUtil.log(slice1);
    }
}

• 池化思想 - 可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能
• 读写指针分离，不需要像 ByteBuffer 一样切换读写模式
• 可以自动扩容
• 支持链式调用，使用更流畅
• 很多地方体现零拷贝，例如
  • slice、duplicate、CompositeByteBuf

三 进阶部分

3.1 半包与粘包问题

3.1.1 现象

• (1）服务端分两次读取到了两个独立的数据包，分别是D1和D2，没有粘包和拆包。
• (2）服务端一次接收到了两个数据包，D1和D2粘合在一起，被称为TCP粘包。
• ( 3）服务端分两次读取到了两个数据包，第一次读取到了完整的D1包和D2包的部分内容，第二次读取到了D2包的剩余内容，这被称为TCP拆包。
• (4)服务端分两次读取到了两个数据包，第一次读取到了D1包的部分内容Dl_1，第二次读取到了D1包的剩余内容D1_2和 D2包的整包。
• 如果此时服务端TCP接收滑窗非常小，而数据包DI和 D2比较大，很有可能会发生第五种可能，即服务端分多次才能将D1和 D2包接收完全，期间发生多次拆包。

服务端

package com.shu.Package;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author shu
 * @Date: 2022/03/03/ 10:27
 * @Description 粘包半包现象
 **/
public class PackageServer {
    static final Logger log = LoggerFactory.getLogger(PackageServer.class);
    void start() {
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            // 半包现象
            serverBootstrap.option(ChannelOption.SO_RCVBUF,10);
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.group(boss, worker);
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            // 连接建立时会执行该方法
                            log.debug("connected {}", ctx.channel());
                            super.channelActive(ctx);
                        }

                        @Override
                        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                            // 连接断开时会执行该方法
                            log.debug("disconnect {}", ctx.channel());
                            super.channelInactive(ctx);
                        }
                    });
                }
            });
            ChannelFuture channelFuture = serverBootstrap.bind(8080);
            log.debug("{} binding...", channelFuture.channel());
            channelFuture.sync();
            log.debug("{} bound...", channelFuture.channel());
            // 关闭channel
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("server error", e);
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
            log.debug("stopped");
        }
    }

    public static void main(String[] args) {
        new PackageServer().start();
    }
}

客服端

package com.shu.Package;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author shu
 * @Date: 2022/03/03/ 10:29
 * @Description 粘包半包现象
 **/
public class PackageClient {
    static final Logger log = LoggerFactory.getLogger(PackageClient.class);
    public static void main(String[] args) {
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(worker);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    log.debug("connected...");
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            log.debug("sending...");
                            // 每次发送16个字节的数据，共发送10次
                            for (int i = 0; i < 10; i++) {
                                ByteBuf buffer = ctx.alloc().buffer();
                                buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
                                ctx.writeAndFlush(buffer);
                            }
                        }
                    });
                }
            });
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
            channelFuture.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            log.error("client error", e);
        } finally {
            worker.shutdownGracefully();
        }
    }
}

粘包现象

• 客服端分十次发送数据，服务端一次接受完毕

半包现象

• 数据接受不完整

3.1.2 原因

• (1）应用程序write 写入的字节大小大于套接口发送缓冲区大小。
• (2）进行MSS大小的TCP分段。
• (3) 以太网帧的payload 大于MTU进行IP分片。

3.1.3 解决方法

• 由于底层的TCP无法理解上层的业务数据，所以在底层是无法保证数据包不被拆分和重组的，这个问题只能通过上层的应用协议栈设计来解决，根据业界的主流协议的解决方案，可以归纳如下。
• (1)消息定长，例如每个报文的大小为固定长度200字节，如果不够，空位补空格;
• (2)在包尾增加回车换行符进行分割，例如FTP协议;
• (3将消息分为消息头和消息体，消息头中包含表示消息总长度（或者消息体长度)的字段，通常设计思路为消息头的第一个字段使用int32来表示消息的总长度;
• (4)更复杂的应用层协议。

3.1.3.1 短连接

• 思路：发送一次数据，关闭连接，重复动作

3.1.3.2 定长解码器

• 客户端于服务器约定一个最大长度，保证客户端每次发送的数据长度都不会大于该长度。
• 若发送数据长度不足则需要补齐至该长度。
• 服务器接收数据时，将接收到的数据按照约定的最大长度进行拆分，即使发送过程中产生了粘包，也可以通过定长解码器将数据正确地进行拆分。
• 服务端需要用到FixedLengthFrameDecoder对数据进行定长解码。

服务端

package com.shu.Package;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author shu
 * @Date: 2022/03/03/ 10:27
 * @Description 粘包半包现象
 **/
public class PackageServer {
    static final Logger log = LoggerFactory.getLogger(PackageServer.class);
    void start() {
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.option(ChannelOption.SO_RCVBUF,10);
            serverBootstrap.group(boss, worker);
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    // 定长解码器
                    ch.pipeline().addLast(new FixedLengthFrameDecoder(16));
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            // 连接建立时会执行该方法
                            log.debug("connected {}", ctx.channel());
                            super.channelActive(ctx);
                        }

                        @Override
                        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                            // 连接断开时会执行该方法
                            log.debug("disconnect {}", ctx.channel());
                            super.channelInactive(ctx);
                        }
                    });
                }
            });
            ChannelFuture channelFuture = serverBootstrap.bind(8080);
            log.debug("{} binding...", channelFuture.channel());
            channelFuture.sync();
            log.debug("{} bound...", channelFuture.channel());
            // 关闭channel
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("server error", e);
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
            log.debug("stopped");
        }
    }

    public static void main(String[] args) {
        new PackageServer().start();
    }
}

客服端

package com.shu.Package;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author shu
 * @Date: 2022/03/03/ 10:29
 * @Description 粘包半包现象
 **/
public class PackageClient {
    static final Logger log = LoggerFactory.getLogger(PackageClient.class);
    public static void main(String[] args) {
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(worker);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    log.debug("connected...");
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            log.debug("sending...");
                            // 约定最大长度为16
                            final int maxLength = 16;
                            // 被发送的数据
                            char c = 'a';
                            // 向服务器发送10个报文
                            for (int i = 0; i < 10; i++) {
                                ByteBuf buffer = ctx.alloc().buffer(maxLength);
                                // 定长byte数组，未使用部分会以0进行填充
                                byte[] bytes = new byte[maxLength];
                                // 生成长度为0~15的数据
                                for (int j = 0; j < (int)(Math.random()*(maxLength-1)); j++) {
                                    bytes[j] = (byte) c;
                                }
                                buffer.writeBytes(bytes);
                                c++;
                                // 将数据发送给服务器
                                ctx.writeAndFlush(buffer);
                            }
                        }
                    });
                }
            });
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
            channelFuture.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            log.error("client error", e);
        } finally {
            worker.shutdownGracefully();
        }
    }
}

3.1.3.3 行解码器

• 行解码器的是通过分隔符对数据进行拆分来解决粘包半包问题的。
• 可以通过LineBasedFrameDecoder(int maxLength)来拆分以换行符(\n)为分隔符的数据，也可以通过DelimiterBasedFrameDecoder(int maxFrameLength, ByteBuf… delimiters)来指定通过什么分隔符来拆分数据（可以传入多个分隔符）。
• 两种解码器都需要传入数据的最大长度，若超出最大长度，会抛出TooLongFrameException异常，以换行符 \n 为分隔符。

服务端

package com.shu.Package;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Author shu
 * @Date: 2022/03/03/ 10:27
 * @Description 粘包半包现象
 **/
public class PackageServer {
    static final Logger log = LoggerFactory.getLogger(PackageServer.class);
    void start() {
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.group(boss, worker);
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    // 基于/n的帧解码器
                    ch.pipeline().addLast(new LineBasedFrameDecoder(64));
                    // 基于自定义字符解码器
                    // 将分隔符放入ByteBuf中
			ByteBuf bufSet = 			ch.alloc().buffer().writeBytes("\\c".getBytes(StandardCharsets.UTF_8));
					// 通过行解码器对粘包数据进行拆分，以 \c 为分隔符
					ch.pipeline().addLast(new DelimiterBasedFrameDecoder(64, 					 ch.alloc().buffer().writeBytes(bufSet)));
                    
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            // 连接建立时会执行该方法
                            log.debug("connected {}", ctx.channel());
                            super.channelActive(ctx);
                        }

                        @Override
                        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                            // 连接断开时会执行该方法
                            log.debug("disconnect {}", ctx.channel());
                            super.channelInactive(ctx);
                        }
                    });
                }
            });
            ChannelFuture channelFuture = serverBootstrap.bind(8080);
            log.debug("{} binding...", channelFuture.channel());
            channelFuture.sync();
            log.debug("{} bound...", channelFuture.channel());
            // 关闭channel
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("server error", e);
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
            log.debug("stopped");
        }
    }

    public static void main(String[] args) {
        new PackageServer().start();
    }
}

客服端

package com.shu.Package;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.Random;

/**
 * @Author shu
 * @Date: 2022/03/03/ 10:29
 * @Description 粘包半包现象
 **/
public class PackageClient {
    static final Logger log = LoggerFactory.getLogger(PackageClient.class);
    public static void main(String[] args) {
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(worker);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    log.debug("connected...");
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            log.debug("sending...");
                            // 约定最大长度为 64
                            final int maxLength = 64;
                            // 被发送的数据
                            char c = 'a';
                            for (int i = 0; i < 10; i++) {
                                ByteBuf buffer = ctx.alloc().buffer(maxLength);
                                // 生成长度为0~62的数据
                                Random random = new Random();
                                StringBuilder sb = new StringBuilder();
                                for (int j = 0; j < (int)(random.nextInt(maxLength-2)); j++) {
                                    sb.append(c);
                                }
                                // 数据以 \n 结尾
                                sb.append("\n");
                                buffer.writeBytes(sb.toString().getBytes(StandardCharsets.UTF_8));
                                c++;
                                // 将数据发送给服务器
                                ctx.writeAndFlush(buffer);
                            }
                            }

                    });
                }
            });
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
            channelFuture.channel().closeFuture().sync();

        } catch (InterruptedException e) {
            log.error("client error", e);
        } finally {
            worker.shutdownGracefully();
        }
    }
}

3.1.3.4 长度字段解码器

在传送数据时可以在数据中添加一个用于表示有用数据长度的字段，在解码时读取出这个用于表明长度的字段，同时读取其他相关参数，即可知道最终需要的数据是什么样子的
LengthFieldBasedFrameDecoder解码器可以提供更为丰富的拆分方法，其构造方法有五个参数

public LengthFieldBasedFrameDecoder(    
 int maxFrameLength,    
 int lengthFieldOffset, 
int lengthFieldLength,    
 int lengthAdjustment, 
int initialBytesToStrip)

参数解析

• maxFrameLength 数据最大长度
  • 表示数据的最大长度（包括附加信息、长度标识等内容）
• lengthFieldOffset 数据长度标识的起始偏移量
  • 用于指明数据第几个字节开始是用于标识有用字节长度的，因为前面可能还有其他附加信息
• lengthFieldLength 数据长度标识所占字节数（用于指明有用数据的长度）
  • 数据中用于表示有用数据长度的标识所占的字节数
• lengthAdjustment 长度表示与有用数据的偏移量
  • 用于指明数据长度标识和有用数据之间的距离，因为两者之间还可能有附加信息
• initialBytesToStrip 数据读取起点
  • 读取起点，不读取 0 ~ initialBytesToStrip 之间的数据

参数图解

详细解释请查看源码文档

public class EncoderStudy {
    public static void main(String[] args) {
        // 模拟服务器
        // 使用EmbeddedChannel测试handler
        EmbeddedChannel channel = new EmbeddedChannel(
                // 数据最大长度为1KB，长度标识前后各有1个字节的附加信息，长度标识长度为4个字节（int）
                new LengthFieldBasedFrameDecoder(1024, 1, 4, 1, 0),
                new LoggingHandler(LogLevel.DEBUG)
        );

        // 模拟客户端，写入数据
        ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
        send(buffer, "Hello");
        channel.writeInbound(buffer);
        send(buffer, "World");
        channel.writeInbound(buffer);
    }

    private static void send(ByteBuf buf, String msg) {
        // 得到数据的长度
        int length = msg.length();
        byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
        // 将数据信息写入buf
        // 写入长度标识前的其他信息
        buf.writeByte(0xCA);
        // 写入数据长度标识
        buf.writeInt(length);
        // 写入长度标识后的其他信息
        buf.writeByte(0xFE);
        // 写入具体的数据
        buf.writeBytes(bytes);
    }
}

3.2 协议的设计

TCP/IP 中消息传输基于流的方式，没有边界
协议的目的就是划定消息的边界，制定通信双方要共同遵守的通信规则

3.2.1 Redis协议

package com.shu.Protocol;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

/**
 * @Author shu
 * @Date: 2022/03/03/ 14:16
 * @Description
 **/
public class RedisProtocol {
    static final Logger log = LoggerFactory.getLogger(RedisProtocol.class);
    public static void main(String[] args) {
        NioEventLoopGroup group =  new NioEventLoopGroup();
        try {
            ChannelFuture channelFuture = new Bootstrap()
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            // 打印日志
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    // 回车与换行符
                                    final byte[] LINE = {'\r','\n'};
                                    // 获得ByteBuf
                                    ByteBuf buffer = ctx.alloc().buffer();
                                    // 连接建立后，向Redis中发送一条指令，注意添加回车与换行
                                    // set name Nyima
                                    buffer.writeBytes("*3".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("$3".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("set".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("$4".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("name".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("$5".getBytes());
                                    buffer.writeBytes(LINE);
                                    buffer.writeBytes("Nyima".getBytes());
                                    buffer.writeBytes(LINE);
                                    ctx.writeAndFlush(buffer);
                                }

                            });
                        }
                    })
                    .connect(new InetSocketAddress("localhost", 6379));
            channelFuture.sync();
            // 关闭channel
            channelFuture.channel().close().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 关闭group
            group.shutdownGracefully();
        }
    }
}

3.2.2 Http协议

package com.shu.Protocol;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;

import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;

/**
 * @Author shu
 * @Date: 2022/03/03/ 14:18
 * @Description http协议
 **/
public class HttpProtocol {
    static final Logger log = LoggerFactory.getLogger(HttpProtocol.class);

    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();
        new ServerBootstrap()
                .group(group)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        // 作为服务器，使用 HttpServerCodec 作为编码器与解码器
                        ch.pipeline().addLast(new HttpServerCodec());
                        // 服务器只处理HTTPRequest
                        ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) {
                                // 获得请求uri
                                log.debug(msg.uri());

                                // 获得完整响应，设置版本号与状态码
                                DefaultFullHttpResponse response = new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
                                // 设置响应内容
                                byte[] bytes = "<h1>Hello, World!</h1>".getBytes(StandardCharsets.UTF_8);
                                // 设置响应体长度，避免浏览器一直接收响应内容
                                response.headers().setInt(CONTENT_LENGTH, bytes.length);
                                // 设置响应体
                                response.content().writeBytes(bytes);

                                // 写回响应
                                ctx.writeAndFlush(response);
                            }
                        });
                    }
                })
                .bind(8080);
    }
}

3.2.3 自定义协议

3.2.3.1 要素

• 魔数：用来在第一时间判定接收的数据是否为无效数据包
• 版本号：可以支持协议的升级
• 序列化算法：消息正文到底采用哪种序列化反序列化方式
  • 如：json、protobuf、hessian、jdk
• 指令类型：是登录、注册、单聊、群聊… 跟业务相关
• 请求序号：为了双工通信，提供异步能力
• 正文长度
• 消息正文

3.2.3.2 代码

package com.shu.Message;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageCodec;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.List;

/**
 * @Author shu
 * @Date: 2022/03/03/ 16:35
 * @Description 消息编解码
 **/
@Slf4j
public class MessageCodec extends ByteToMessageCodec<Message> {
    @Override
    public void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
        // 1. 4 字节的魔数
        out.writeBytes(new byte[]{1, 2, 3, 4});
        // 2. 1 字节的版本,
        out.writeByte(1);
        // 3. 1 字节的序列化方式 jdk 0 , json 1
        out.writeByte(0);
        // 4. 1 字节的指令类型
        out.writeByte(msg.getMessageType());
        // 5. 4 个字节
        out.writeInt(msg.getSequenceId());
        // 无意义，对齐填充
        out.writeByte(0xff);
        // 6. 获取内容的字节数组
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(bos);
        oos.writeObject(msg);
        byte[] bytes = bos.toByteArray();
        // 7. 长度
        out.writeInt(bytes.length);
        // 8. 写入内容
        out.writeBytes(bytes);
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        int magicNum = in.readInt();
        byte version = in.readByte();
        byte serializerType = in.readByte();
        byte messageType = in.readByte();
        int sequenceId = in.readInt();
        in.readByte();
        int length = in.readInt();
        byte[] bytes = new byte[length];
        in.readBytes(bytes, 0, length);
        ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
        Message message = (Message) ois.readObject();
        log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
        log.debug("{}", message);
        out.add(message);
    }
}

3.3 重要参数

CONNECT_TIMEOUT_MILLIS

• 属于 SocketChannal 的参数
• 用在客户端建立连接时，如果在指定毫秒内无法连接，会抛出 timeout 异常

public class TestParam {
    public static void main(String[] args) {
        // SocketChannel 5s内未建立连接就抛出异常
        new Bootstrap().option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);
        
        // ServerSocketChannel 5s内未建立连接就抛出异常
        new ServerBootstrap().option(ChannelOption.CONNECT_TIMEOUT_MILLIS,5000);
        // SocketChannel 5s内未建立连接就抛出异常
        new ServerBootstrap().childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);
    }
}

SO_BACKLOG

在Netty中，SO_BACKLOG主要用于设置全连接队列的大小。当处理Accept的速率小于连接建立的速率时，全连接队列中堆积的连接数大于SO_BACKLOG设置的值是，便会抛出异常

// 设置全连接队列，大小为2
new ServerBootstrap().option(ChannelOption.SO_BACKLOG, 2);

TCP_NODELAY

• 属于 SocketChannal 参数
• 因为 Nagle 算法，数据包会堆积到一定的数量后一起发送，这就可能导致数据的发送存在一定的延时
• 该参数默认为false，如果不希望的发送被延时，则需要将该值设置为true

SO_SNDBUF & SO_RCVBUF

• SO_SNDBUF 属于 SocketChannal 参数
• SO_RCVBUF 既可用于 SocketChannal 参数，也可以用于 ServerSocketChannal 参数（建议设置到 ServerSocketChannal 上）
• 该参数用于指定接收方与发送方的滑动窗口大小，不过不建议去调整这两个参数，也不应该过大，否则会占用操作系统的缓冲区

ALLOCATOR

• 属于 SocketChannal 参数
• 用来配置 ByteBuf 是池化还是非池化，是直接内存还是堆内存

RCVBUF_ALLOCATOR

• 属于 SocketChannal 参数
• 控制 Netty 接收缓冲区大小
• 负责入站数据的分配，决定入站缓冲区的大小（并可动态调整），统一采用 direct 直接内存，具体池化还是非池化由 allocator 决定