ByteBuf buf = Unpooled.buffer(size).order(ByteOrder.BIG_ENDIAN);

final ChannelPipeline pipeline = channel.pipeline();
 final SslHandler handler = new SslHandler(clientEngine);
 handler.setEnableRenegotiation(true);// 重新协商
 pipeline.addLast("ssl", handler);
 pipeline.addLast("decoder", new ApnsResponseDecoder());

HashedWheelTimer timer = new HashedWheelTimer(100, TimeUnit.MILLISECONDS, 16);
System.out.println(LocalTime.now());
timer.newTimeout((timeout) -> {
    System.out.println(LocalTime.now());
    System.out.println(timeout);
}, 5, TimeUnit.SECONDS);
//阻塞main线程
System.in.read();

-- SslHandler:
public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundHandler 
-- 其中的ByteToMessageDecoder:
public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter

public class SslChannelInitializer extends ChannelInitializer<Channel> {
    private final SslContext context;
    private final boolean startTls;

    public SslChannelInitializer(SslContext context,boolean startTls){
        this.context = context;
        this.startTls = startTls;
    }
    @Override
    protected void initChannel(Channel ch) throws Exception {
        SSLEngine engine = context.newEngine(ch.alloc());
        ch.pipeline().addFirst("ssl"
            , new SslHandler(engine, startTls));
    }
}

public class HttpPipelineInitializer extends ChannelInitializer<Channel> {

    private final boolean client;

    public HttpPipelineInitializer(boolean client) {
        this.client = client;
    }

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        if (client) {
            pipeline.addLast("decoder", new HttpResponseDecoder());  //1
            pipeline.addLast("encoder", new HttpRequestEncoder());  //2
        } else {
            pipeline.addLast("decoder", new HttpRequestDecoder());  //3
            pipeline.addLast("encoder", new HttpResponseEncoder());  //4
        }
    }
}

public class HttpAggregatorInitializer extends ChannelInitializer<Channel> {

    private final boolean client;

    public HttpAggregatorInitializer(boolean client) {
        this.client = client;
    }

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        if (client) {
            pipeline.addLast("codec", new HttpClientCodec());  //1
        } else {
            pipeline.addLast("codec", new HttpServerCodec());  //2
        }
        pipeline.addLast("aggegator", new HttpObjectAggregator(512 * 1024));  //3 512kb
    }
}

public class HttpAggregatorInitializer extends ChannelInitializer<Channel> {

    private final boolean isClient;
    public HttpAggregatorInitializer(boolean isClient) {
        this.isClient = isClient;
    }
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        if (isClient) {
            pipeline.addLast("codec", new HttpClientCodec()); //1
            pipeline.addLast("decompressor",new HttpContentDecompressor()); //2
        } else {
            pipeline.addLast("codec", new HttpServerCodec()); //3
            pipeline.addLast("compressor",new HttpContentCompressor()); //4
        }
    }
}

public class HttpsCodecInitializer extends ChannelInitializer<Channel> {

    private final SslContext context;
    private final boolean client;

    public HttpsCodecInitializer(SslContext context, boolean client) {
        this.context = context;
        this.client = client;
    }

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        SSLEngine engine = context.newEngine(ch.alloc());
        pipeline.addFirst("ssl", new SslHandler(engine));  //1

        if (client) {
            pipeline.addLast("codec", new HttpClientCodec());  //2
        } else {
            pipeline.addLast("codec", new HttpServerCodec());  //3
        }
    }
}

BinaryWebSocketFrame: 二进制;
TextWebSocketFrame: 文本;
ContunuationWebSocketFrame: 后续数据;

PingWebSocketFrame: ping，对方会回pong;
PongWebSocketFrame: pong;
CloseWebSocketFrame: 关闭。

public class WebSocketServerInitializer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel ch) throws Exception {
        ch.pipeline().addLast(
            new HttpServerCodec(),
            //为握手提供聚合的 HttpRequest
            new HttpObjectAggregator(65536),
            //如果被请求的端点是"/websocket"，则处理该升级握手
            new WebSocketServerProtocolHandler("/websocket"),
            //TextFrameHandler 处理 TextWebSocketFrame
            new TextFrameHandler(),
            //BinaryFrameHandler 处理 BinaryWebSocketFrame
            new BinaryFrameHandler(),
            //ContinuationFrameHandler 处理 ContinuationWebSocketFrame
            new ContinuationFrameHandler());
    }

    public static final class TextFrameHandler extends
        SimpleChannelInboundHandler<TextWebSocketFrame> {
        @Override
        public void channelRead0(ChannelHandlerContext ctx,
            TextWebSocketFrame msg) throws Exception {
            // Handle text frame
        }
    }

    public static final class BinaryFrameHandler extends
        SimpleChannelInboundHandler<BinaryWebSocketFrame> {
        @Override
        public void channelRead0(ChannelHandlerContext ctx,
            BinaryWebSocketFrame msg) throws Exception {
            // Handle binary frame
        }
    }

    public static final class ContinuationFrameHandler extends
        SimpleChannelInboundHandler<ContinuationWebSocketFrame> {
        @Override
        public void channelRead0(ChannelHandlerContext ctx,
            ContinuationWebSocketFrame msg) throws Exception {
            // Handle continuation frame
        }
    }
}

FileInputStream in = new FileInputStream(file);
//以该文件的完整长度创建一个新的 DefaultFileRegion
FileRegion region = new DefaultFileRegion(
        in.getChannel(), 0, file.length());
//发送该 DefaultFileRegion，并注册一个 ChannelFutureListener
channel.writeAndFlush(region).addListener(
    new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future)
       throws Exception {
       if (!future.isSuccess()) {
           //处理失败
           Throwable cause = future.cause();
           // Do something
       }
    }
});

@Override
protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    //将 SslHandler 添加到 ChannelPipeline 中
    pipeline.addLast(new SslHandler(sslCtx.newEngine(ch.alloc())));
    //添加 ChunkedWriteHandler 以处理作为 ChunkedInput 传入的数据
    pipeline.addLast(new ChunkedWriteHandler());
    //一旦连接建立，WriteStreamHandler 就开始写文件数据
    pipeline.addLast(new WriteStreamHandler());
}

public final class WriteStreamHandler
extends ChannelInboundHandlerAdapter {
@Override
//当连接建立时，channelActive() 方法将使用 ChunkedInput 写文件数据
    public void channelActive(ChannelHandlerContext ctx)
        throws Exception {
        super.channelActive(ctx);
        ctx.writeAndFlush(
        new ChunkedStream(new FileInputStream(file))); // ChunkedStream => ChunkedInput
    }
}

public class MarshallingInitializer extends ChannelInitializer<Channel> {
    private final MarshallerProvider marshallerProvider;
    private final UnmarshallerProvider unmarshallerProvider;

    public MarshallingInitializer(
            UnmarshallerProvider unmarshallerProvider,
            MarshallerProvider marshallerProvider) {
        this.marshallerProvider = marshallerProvider;
        this.unmarshallerProvider = unmarshallerProvider;
    }

    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        //添加 MarshallingDecoder 以将 ByteBuf 转换为 POJO
        pipeline.addLast(new MarshallingDecoder(unmarshallerProvider));
        //添加 MarshallingEncoder 以将POJO 转换为 ByteBuf
        pipeline.addLast(new MarshallingEncoder(marshallerProvider));
        //添加 ObjectHandler，以处理普通的实现了Serializable 接口的 POJO
        pipeline.addLast(new ObjectHandler());
    }

    public static final class ObjectHandler
        extends SimpleChannelInboundHandler<Serializable> {
        @Override
        public void channelRead0(
            ChannelHandlerContext channelHandlerContext,
            Serializable serializable) throws Exception {
            // Do something
        }
    }
}

public static MarshallingDecoder buildMarshallingDecoder() {
       //首先通过Marshalling工具类的getProvidedMarshallerFactory静态方法获取MarshallerFactory实例
       //参数“serial”表示创建的是Java序列化工厂对象，它由jboss-marshalling-serial-1.3.0.CR9.jar提供。
       final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial");
       //创建了MarshallingConfiguration对象
       final MarshallingConfiguration configuration = new MarshallingConfiguration();
       //将它的版本号设置为5
       configuration.setVersion(5);
       //然后根据MarshallerFactory和MarshallingConfiguration创建UnmarshallerProvider实例
       UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration);
       //最后通过构造函数创建Netty的MarshallingDecoder对象
       //它有两个参数，分别是UnmarshallerProvider和单个消息序列化后的最大长度。
       MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024);
       return decoder;
   }

// 解码:
ProtobufVarint32FrameDecoder: bytes=>msg; 解析出头部的长度字段,以正确划分帧;
ProtobufDecoder: msg=>msg;
// 编码:
ProtobufVarint32LengthFieldPrepender: msg=>bytes; 头部添加长度字段.
ProtobufEncoder: msg=>msg.

.childHandler(new ChannelInitializer<SocketChannel>() {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline()
                .addLast(new ProtobufVarint32FrameDecoder())
                .addLast(new ProtobufDecoder(
                        ProtoObject.Req.getDefaultInstance()))
                .addLast(new ProtobufVarint32LengthFieldPrepender())
                .addLast(new ProtobufEncoder())
                .addLast(new ServerHandler());
    }
})

.childHandler(new ChannelInitializer<SocketChannel>() {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline()
                .addLast(new ProtobufVarint32FrameDecoder())
                .addLast(new ProtobufDecoder(
                        ProtoObject.Resp.getDefaultInstance()))
                .addLast(new ProtobufVarint32LengthFieldPrepender())
                .addLast(new ProtobufEncoder())
                .addLast(new ClientHandler());
    }
})

public interface Message extends MessageLite, MessageOrBuilder {...}

// 1.字节=>消息: （解码）
ByteToMessageDecoder: 抽象类;
LineBasedFrameDecoder: 行分割消息数据， 实际类。
HttpObjectDecoder: Http数据解析。抽象类。
// 2.消息=>消息: （格式转换）
MessageToMessageDecoder: 消息=>消息;
HttpObjectAggregator: Http数据转换。实际类。

public class LineBasedFrameDecoder extends ByteToMessageDecoder;

public abstract class HttpObjectDecoder extends ByteToMessageDecoder;

public class HttpObjectAggregator extends MessageAggregator<HttpObject, HttpMessage, HttpContent, FullHttpMessage>;

public abstract class MessageAggregator<I, S, C extends ByteBufHolder, O extends ByteBufHolder> extends MessageToMessageDecoder<I>;

public class ToIntegerDecoder extends ByteToMessageDecoder {
    @Override
    public void decode(ChannelHandlerContext ctx
        ,ByteBuf in         // 1.输入
        ,List<Object> out)  // 2.输出
        throws Exception {
        //检查是否至少有 4 字节可读（一个 int 的字节长度）
        if (in.readableBytes() >= 4) {
            out.add(in.readInt());
        }
    }
}

decode:     必须实现，解析每条消息;
decodeLast: 可选，处理最后一条消息，默认是调用decode。

public class ToIntegerDecoder2 extends ReplayingDecoder<Void> {
    @Override
    public void decode(ChannelHandlerContext ctx, ByteBuf in, 
    //传入的 ByteBuf 是 ReplayingDecoderByteBuf
        List<Object> out) throws Exception {
        //从入站 ByteBuf 中读取 一个 int，并将其添加到解码消息的 List 中
        out.add(in.readInt());
    }
}

public abstract class MessageToMessageDecoder<I>
extends ChannelInboundHandlerAdapter

public class IntegerToStringDecoder extends
    MessageToMessageDecoder<Integer> {
    @Override
    public void decode(ChannelHandlerContext ctx
        , Integer msg      // 1.输入
        , List<Object> out // 2.输出
) throws Exception {
        out.add(String.valueOf(msg));
    }
}

public class SafeByteToMessageDecoder extends ByteToMessageDecoder {
    private static final int MAX_FRAME_SIZE = 1024;
    @Override
    public void decode(ChannelHandlerContext ctx, ByteBuf in,
        List<Object> out) throws Exception {
            int readable = in.readableBytes();
            if (readable > MAX_FRAME_SIZE) {
                in.skipBytes(readable); // 一种可能的处理
                throw new TooLongFrameException("Frame too big!");// 抛出异常
        }
    }
}

MessageToByteEncoder<I>: 消息=>字节
MessageToMessageEncoder: 消息=>消息
Websocket08FrameEncoder: 消息=>消息
ProtobufEncoder: 消息=>消息

public class IntegerToStringEncoder
    extends MessageToMessageEncoder<Integer> {
    @Override
    public void encode(ChannelHandlerContext ctx
        , Integer msg       // 1.输入 
        , List<Object> out  // 2.输出
) throws Exception {
        out.add(String.valueOf(msg));
    }
}

decode(ctx,ByteBuf in,List<Object>out)
decodeLast(ctx,ByteBuf in,List<Object>out)
encode(ctx,I msg,ByteBuf out)

public abstract class MessageToMessageCodec<INBOUND_IN, OUTBOUND_IN> extends ChannelDuplexHandler
public class ChannelDuplexHandler extends ChannelInboundHandlerAdapter implements ChannelOutboundHandler

INBOUND_IN:  入站数据参数的数据类型
OUTBOUND_IN: 出站数据参数的数据类型

protected abstract void encode(ChannelHandlerContext ctx
, OUTBOUND_IN msg   -- 输入
, List<Object> out) -- 输出，实际上一般会是INBOUND_IN类型
throws Exception;

protected abstract void decode(ChannelHandlerContext ctx
, INBOUND_IN msg    -- 输入
, List<Object> out) -- 输出，实际上一般会是OUTBOUND_IN类型
throws Exception;

public class CombinedChannelDuplexHandler<I extends ChannelInboundHandler
, O extends ChannelOutboundHandler>
        extends ChannelDuplexHandler {

public class CombinedByteCharCodec extends
    CombinedChannelDuplexHandler<ByteToCharDecoder, CharToByteEncoder> {
    public CombinedByteCharCodec() {
        //将委托实例传递给父类
        super(new ByteToCharDecoder(), new CharToByteEncoder());
    }
}

@Test
public void testFramesDecoded() {
    //创建一个 ByteBuf，并存储 9 字节
    ByteBuf buf = Unpooled.buffer();
    for (int i = 0; i < 9; i++) {
        buf.writeByte(i);
    }
    ByteBuf input = buf.duplicate(); // 1
    EmbeddedChannel channel = new EmbeddedChannel(
        new FixedLengthFrameDecoder(3));
     
    assertTrue(channel.writeInbound(input.retain())); // 2. 数据量够,能够传到末端
    assertTrue(channel.finish()); // 3. 还有可读数据,返回true
   
    // 读3帧(3个入站消息):
    ByteBuf read = (ByteBuf) channel.readInbound();
    assertEquals(buf.readSlice(3), read); // 4
    read.release(); // 5

    read = (ByteBuf) channel.readInbound();
    assertEquals(buf.readSlice(3), read);
    read.release();

    read = (ByteBuf) channel.readInbound();
    assertEquals(buf.readSlice(3), read);
    read.release();

    assertNull(channel.readInbound());
    buf.release();
}

try {
        channel.writeInbound(input.readBytes(4));
    } catch (TooLongFrameException e) {
    Assert.fail();// 期望不能抵达这里
    }
}

try {
            channel.writeInbound(input.readBytes(4));
            Assert.fail();// 期望不能抵达这里。
        } catch (TooLongFrameException e) {
        }
}

public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C extends Channel> implements Cloneable {}

public class Bootstrap extends AbstractBootstrap<Bootstrap, Channel> {}

public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {}

public B channel(Class<? extends C> channelClass) {// 设置Channel
        if (channelClass == null) {
            throw new NullPointerException("channelClass");
        }
        return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
    }
public B group(EventLoopGroup group);// 设置EventLoop实现
public B localAddress(SocketAddress localAddress) {// 设置本地地址
        this.localAddress = localAddress;
        return (B) this;
    }
public B localAddress(String inetHost, int inetPort) {// 设置本地地址
        return localAddress(SocketUtils.socketAddress(inetHost, inetPort));
    }
public <T> B option(ChannelOption<T> option, T value) { // 设置全局ChannelOption
        if (option == null) {
            throw new NullPointerException("option");
        }
        if (value == null) {
            synchronized (options) {
                options.remove(option);
            }
        } else {
            synchronized (options) {
                options.put(option, value);
            }
        }
        return (B) this;
  }
public <T> B attr(AttributeKey<T> key, T value);// 设置Channel的属性，后文会具体提到

public B handler(ChannelHandler handler)；// 设置添加到pipeline的handler

public ChannelFuture bind(); // 绑定channel
public ChannelFuture connect(); // 连接到远程节点

public void bootstrap() {
        //设置 EventLoopGroup，提供用于处理 Channel 事件的 EventLoop
        EventLoopGroup group = new NioEventLoopGroup();
        //创建一个Bootstrap类的实例以创建和连接新的客户端Channel
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
            .channel(NioSocketChannel.class) // NIO
            .handler(new SimpleChannelInboundHandler<ByteBuf>() {
                @Override
                protected void channelRead0(
                    ChannelHandlerContext channelHandlerContext,
                    ByteBuf byteBuf) throws Exception {
                    System.out.println("Received data");
                }
                });
        //连接到远程主机
        ChannelFuture future = bootstrap.connect(
            new InetSocketAddress("xxx.com", 80));
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture)
                throws Exception {
                if (channelFuture.isSuccess()) {
                    System.out.println("Connection established");
                } else {
                    System.err.println("Connection attempt failed");
                    channelFuture.cause().printStackTrace();
                }
            }
        });
    }

childOption: 后续accept的子Channel的配置; // bind后生效
childAttr: 设置给 已经accept 的子Channel属性; 
childHanlder: 设置给 已经accept 的子Channel的pipeline。

option: 后续新建的serverChannel的配置; // bind后生效
attr: 设置给当前serverChannel的属性;
handler: 设置给当前serverChannel的pipeline。

/**
     * 代码清单 8-4 引导服务器
     * */
    public void bootstrap() {
        NioEventLoopGroup group = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();
        // 1. 设置:
        bootstrap.group(group)
            .channel(NioServerSocketChannel.class)
            .childHandler(new SimpleChannelInboundHandler<ByteBuf>() {
                @Override
                protected void channelRead0(ChannelHandlerContext channelHandlerContext,
                    ByteBuf byteBuf) throws Exception {
                    System.out.println("Received data");
                }
            });
        // 2. bind: 通过配置好的 ServerBootstrap 的实例绑定该 Channel
        ChannelFuture future = bootstrap.bind(new InetSocketAddress(8080));
        // 3. listener future:
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture)
                throws Exception {
                if (channelFuture.isSuccess()) {
                    System.out.println("Server bound");
                } else {
                    System.err.println("Bind attempt failed");
                    channelFuture.cause().printStackTrace();
                }
            }
        });
    }

public interface EventLoop extends OrderedEventExecutor, EventLoopGroup {
    @Override
    EventLoopGroup parent();
}

public void bootstrap() {
        ServerBootstrap bootstrap = new ServerBootstrap();
        // 1. 配置:
        bootstrap.group(new NioEventLoopGroup(), new NioEventLoopGroup())
            .channel(NioServerSocketChannel.class)
            .childHandler(// 注意是childHandler
                new SimpleChannelInboundHandler<ByteBuf>() {
                    ChannelFuture connectFuture; // handler的成员变量
                    @Override
                    public void channelActive(ChannelHandlerContext ctx)
                        throws Exception {
                        // 1.1 新建boostrap
                        Bootstrap bootstrap = new Bootstrap();
                        bootstrap.channel(NioSocketChannel.class).handler(
                        // 注意是Handler
                            new SimpleChannelInboundHandler<ByteBuf>() {
                                @Override
                                protected void channelRead0(
                                    ChannelHandlerContext ctx, ByteBuf in)
                                    throws Exception {
                                    System.out.println("Received data");
                                }
                            });
                        // 1.2 bootstrap配置: 使用channel的eventLoop引用:
                        bootstrap.group(ctx.channel().eventLoop());
                        connectFuture = bootstrap.connect(
                            new InetSocketAddress("xxx.com", 80));
                    }
                    @Override
                    protected void channelRead0(
                        ChannelHandlerContext channelHandlerContext,
                            ByteBuf byteBuf) throws Exception {
                        if (connectFuture.isDone()) { 
                            // 1.3: 当连接完成时，执行一些数据操作（如代理）
                        }
                    }
                });
        
        // 2. bind: 
        ChannelFuture future = bootstrap.bind(new InetSocketAddress(8080));
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture)
                throws Exception {
                if (channelFuture.isSuccess()) {
                    System.out.println("Server bound");
                } else {
                    System.err.println("Bind attempt failed");
                    channelFuture.cause().printStackTrace();
                }
            }
        });
    }

bootstrap.group(new NioEventLoopGroup(), new NioEventLoopGroup())

@Override
public ServerBootstrap group(EventLoopGroup group) {
    return group(group, group);
}
 
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
    super.group(parentGroup);
    if (childGroup == null) {
        throw new NullPointerException("childGroup");
    }
    if (this.childGroup != null) {
        throw new IllegalStateException("childGroup set already");
    }
    this.childGroup = childGroup;
    return this;
}

bootstrap.option(ChannelOption.SO_KEEPALIVE, true)
         .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);

// 1. 声明:
final AttributeKey<Integer> id = AttributeKey.newInstance("ID");
Bootstrap bootstrap = new Bootstrap();
// 2. 赋值:
bootstrap.attr(id, 123456);
// 2. 使用:(某个handler里头:)
@Override
public void channelRegistered(ChannelHandlerContext ctx)
    throws Exception {
    Integer idValue = ctx.channel().attr(id).get();
    // do something with the idValue
}

// 1. OIO版本:
Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(new OioEventLoopGroup()).channel(
            OioDatagramChannel.class).handler(
            new SimpleChannelInboundHandler<DatagramPacket>() {
                @Override
                public void channelRead0(ChannelHandlerContext ctx,
                    DatagramPacket msg) throws Exception {
                    // Do something with the packet
                }
            }
        );
ChannelFuture future = bootstrap.bind(new InetSocketAddress(0));

// 2. NIO版本
// 服务端:
final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioDatagramChannel.class);
bootstrap.group(nioEventLoopGroup);
bootstrap.handler(new ChannelInitializer<NioDatagramChannel>() {...});
ChannelFuture sync = bootstrap.bind(9009).sync();
Channel udpChannel = sync.channel();
sync.closeFuture().await();// 等待关闭。
// 客户端:
final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.channel(NioDatagramChannel.class);
bootstrap.group(nioEventLoopGroup);
bootstrap.handler(new ChannelInitializer<NioDatagramChannel>() {...});
ChannelFuture sync = bootstrap.bind(0).sync();
Channel udpChannel = sync.channel();
String data = "data";
udpChannel.writeAndFlush(new DatagramPacket(Unpooled.copiedBuffer(data.getBytes(Charset.forName("UTF-8"))), new InetSocketAddress("192.168.2.29", 9008)));
sync.closeFuture().await();// 等待关闭。

 //shutdownGracefully()方法将释放所有的资源，并且关闭所有的当前正在使用中的 Channel
Future<?> future = group.shutdownGracefully();
// block until the group has shutdown
future.syncUninterruptibly();

public static void executeTaskInEventLoop() {
       boolean terminated = true;
       //...
       while (!terminated) {
           //阻塞，直到有事件已经就绪可被运行
           List<Runnable> readyEvents = blockUntilEventsReady();
           for (Runnable ev: readyEvents) {
               //循环遍历，并处理所有的事件
               ev.run();
           }
       }
   }

public static void schedule() {
    // 线程池:
    ScheduledExecutorService executor =
            Executors.newScheduledThreadPool(10);
    ScheduledFuture<?> future = executor.schedule(
        new Runnable() {
        @Override
        public void run() {
            System.out.println("Now it is 60 seconds later");
        }
    }, 60, TimeUnit.SECONDS); 
    executor.shutdown();
}

public static void scheduleViaEventLoop() {
    Channel ch = CHANNEL_FROM_SOMEWHERE; // get reference from somewhere
    ScheduledFuture<?> future = ch.eventLoop().schedule(
        new Runnable() {
        @Override
        public void run() {
            System.out.println("60 seconds later");
        }
    }, 60, TimeUnit.SECONDS); // 60s后
}

public static void scheduleFixedViaEventLoop() {
    Channel ch = CHANNEL_FROM_SOMEWHERE; // get reference from somewhere
    ScheduledFuture<?> future = ch.eventLoop().scheduleAtFixedRate(
       new Runnable() {
       @Override
       public void run() {
            System.out.println("Run every 60 seconds");
       }
    }, 60, 60, TimeUnit.SECONDS);
}

if(当前线程 in 匹配EventLoop中的线程) 直接执行;
else{
    任务放入内部队列。
}

Channel.config.setWriteHighWaterMark();
Channel.config.setWriteLowWaterMark();

bufferBuf.release();

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    //通过调用 ReferenceCountUtil.release()方法释放资源
    ReferenceCountUtil.release(msg);
}
// 底层调的是这个:
public static boolean release(Object msg) {
        if (msg instanceof ReferenceCounted) {
            return ((ReferenceCounted) msg).release();
        }
        return false;
}

@Override
public void write(ChannelHandlerContext ctx,
    Object msg, ChannelPromise promise) {
    ReferenceCountUtil.release(msg);
    //通知 ChannelPromise数据已经被处理了
    promise.setSuccess();
}

ChannelPromise(子接口) -> ChannelFuture(父接口)

Uncompleted => success/fail/cancelled

// 方法1:
io.netty.channel.ChannelFuture future = channel.write(someMessage);
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(io.netty.channel.ChannelFuture f) {
                if (!f.isSuccess()) {
                    f.cause().printStackTrace();
                    f.channel().close();
                }
            }
        });
// 方法2:
public class OutboundExceptionHandler extends ChannelOutboundHandlerAdapter {
    @Override
    public void write(ChannelHandlerContext ctx, Object msg,
        ChannelPromise promise) {
        promise.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture f) {
                if (!f.isSuccess()) {
                    f.cause().printStackTrace();
                    f.channel().close();
                }
            }
        });
    }
}

public static void modifyPipeline() {
    ChannelPipeline pipeline = CHANNEL_PIPELINE_FROM_SOMEWHERE; // get reference to pipeline;
    //创建一个 FirstHandler 的实例
    FirstHandler firstHandler = new FirstHandler();
    //将该实例作为"handler1"添加到ChannelPipeline 中
    pipeline.addLast("handler1", firstHandler);
    //将一个 SecondHandler的实例作为"handler2"添加到 ChannelPipeline的第一个槽中。这意味着它将被放置在已有的"handler1"之前
    pipeline.addFirst("handler2", new SecondHandler());
    //将一个 ThirdHandler 的实例作为"handler3"添加到 ChannelPipeline 的最后一个槽中
    pipeline.addLast("handler3", new ThirdHandler());
    //...
    //通过名称移除"handler3"
    pipeline.remove("handler3");
    //通过引用移除FirstHandler（它是唯一的，所以不需要它的名称）
    pipeline.remove(firstHandler);
    //将 SecondHandler("handler2")替换为 FourthHandler:"handler4"
    pipeline.replace("handler2", "handler4", new FourthHandler());
}

public interface ChannelHandlerContext extends AttributeMap, ChannelInboundInvoker, ChannelOutboundInvoker

@Override
    public void exceptionCaught(ChannelHandlerContext ctx,
        Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

@Override
public void operationComplete(io.netty.channel.ChannelFuture f) {
    if (!f.isSuccess()) {
        f.cause().printStackTrace();
        f.channel().close();
    }
}

private final static Random random = new Random();
private static final ByteBuf BYTE_BUF_FROM_SOMEWHERE = Unpooled.buffer(1024);
private static final Channel CHANNEL_FROM_SOMEWHERE = new NioSocketChannel();
private static final ChannelHandlerContext CHANNEL_HANDLER_CONTEXT_FROM_SOMEWHERE = DUMMY_INSTANCE;
public static void heapBuffer() {
    ByteBuf heapBuf = BYTE_BUF_FROM_SOMEWHERE; //get reference form somewhere
    //检查 ByteBuf 是否有一个支撑数组
    if (heapBuf.hasArray()) {
        //如果有，则获取对该数组的引用
        byte[] array = heapBuf.array();
        //计算第一个字节的偏移量
        int offset = heapBuf.arrayOffset() + heapBuf.readerIndex();
        //获得可读字节数
        int length = heapBuf.readableBytes();
        //使用数组、偏移量和长度作为参数调用你的方法
        handleArray(array, offset, length);
    }
}

public static void directBuffer() {
    ByteBuf directBuf = BYTE_BUF_FROM_SOMEWHERE; //get reference form somewhere
    //检查 ByteBuf 是否由数组支撑。如果不是，则这是一个直接缓冲区
    if (!directBuf.hasArray()) {
        //获取可读字节数
        int length = directBuf.readableBytes();
        //分配一个新的数组来保存具有该长度的字节数据
        byte[] array = new byte[length];
        //将字节复制到该数组
        directBuf.getBytes(directBuf.readerIndex(), array);
        //使用数组、偏移量和长度作为参数调用你的方法
        handleArray(array, 0, length);
    }
}

 public static void byteBufComposite() {
    CompositeByteBuf messageBuf = Unpooled.compositeBuffer();
    ByteBuf headerBuf = BYTE_BUF_FROM_SOMEWHERE; // can be backing or direct
    ByteBuf bodyBuf = BYTE_BUF_FROM_SOMEWHERE;   // can be backing or direct
    //将 ByteBuf 实例追加到 CompositeByteBuf
    messageBuf.addComponents(headerBuf, bodyBuf);
    //...
    //删除位于索引位置为 0（第一个组件）的 ByteBuf
    messageBuf.removeComponent(0); // remove the header
    //循环遍历所有的 ByteBuf 实例
    for (ByteBuf buf : messageBuf) {
        System.out.println(buf.toString());
    }
}

public static void byteBufCompositeArray() {
    CompositeByteBuf compBuf = Unpooled.compositeBuffer();
    //获得可读字节数
    int length = compBuf.readableBytes();
    //分配一个具有可读字节数长度的新数组
    byte[] array = new byte[length];
    //将字节读到该数组中
    compBuf.getBytes(compBuf.readerIndex(), array);
    //使用偏移量和长度作为参数使用该数组
    handleArray(array, 0, array.length);
}

for(int i;i<buffer.capacity();i++){
    byte b=buffer.getByte(i);
}

//迭代缓冲区的可读字节。
 ByteBuf buffer = ...;
 while (buffer.isReadable()) {
     System.out.println(buffer.readByte());
 }

//调用:
byteBuf.discardReadBytes();

// 1.
byteBuf.writeBytes(otherByteBuf);
// 2.
byteBuf.writeInt(26);

markReaderIndex()// 标记为x
resetReaderIndex()// 回到x,注意不是0
markWriterIndex() // 标记为y
resetWriterIndex()// 回到y,注意不是0
clear() // 这个才是双双清0
readerIndex(int)
writerIndex(int)

@Override
    public ByteBuf markReaderIndex() {
        markedReaderIndex = readerIndex;
        return this;
    }

    @Override
    public ByteBuf resetReaderIndex() {
        readerIndex(markedReaderIndex);
        return this;
    }

//1:
int index = byteBuf.indexOf(xxx)
//2.
int index = byteBuf.forEachByte(ByteBufProcessor.FIND_CR);
// ByteBufProcessor 4.1版本改为 io.netty.util.ByteProcessor

duplicate()
slice(int,int)
Unpooled.unmodifiableBuffer(...)
order(ByteOrder)
readSlice(int)

copy(int,int)

ByteBuf content()// 返回持有的ByteBuf
copy()
duplicate()// 浅拷贝

// 1. 从channel:
Channel channel = ...;
ByteBufAllocator allocator = channel.alloc(); //1
// 2. 从ctx:
ChannelHandlerContext ctx = ...;
ByteBufAllocator allocator2 = ctx.alloc(); //2

Buffer()
heapBuffer() 
directBuffer() 
compositeBuffer()
...// 总之就是各种创建ByteBuf的方法

PooledByteBufAllocator  // 池化的,性能最高 
UnpooledByteBufAllocator // 非池化的