Netty

2022-02-05

Netty——异步和事件驱动

Java网络编程

阻塞I/O

ServerSocket serverSocket = new ServerSocket(portNumber); //创建一个新的ServerSocket，用以监听指定端口上的连接请求
Socket clientSocket = serverSocket.accept();  //对accept()方法的调用将被阻塞，直到一个连接建立
BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream())); 
PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true);
String request, response;
while((request = in.readLine()) != null){
    if("Done".equals(request)) break;
    response = processRequest(request); // 请求被传递给服务器的处理方法
    out.println(response);
}

要点：

ServerSocket上的accept()方法将会一直阻塞到一个连接建立，随后返回一个新的Socket用于客户端和服务器之间的通信。该ServerSocket将继续监听传入的连接；
BufferedReader和PrintWriter都衍生自Socket的输入输出流。前者从一个字符输入流中读取文本，后者打印对象的格式化的表示到文本输出流；
readLine()方法将会阻塞，直到在while内一个由换行符或回车符结尾的字符串被读取。

缺点：

在任何时候都可能有大量的线程处于休眠状态，只是等待输入或者输出数据就绪，这可能算是一种资源浪费；
需要为每个线程的调用栈都分配内存，其默认值大小区间为64KB到1MB，具体取决于操作系统；
即使JVM在物理上可以支持非常大数量的线程，但是远在到达该极限之前，上下文切换所带来的的开销就会带来麻烦。

NIO

class java.nio.channels.Selector是Java的非阻塞I/O实现的关键。它使用了事件通知API以去诶定在一组非阻塞套接字中有哪些已经就绪能够进行I/O相关的操作。因为可以在任何的时间检查任意的读操作或者写操作的完成状态，如图所示，一个单一的线程便可以处理多个并发的连接。

总体而言，与阻塞I/O模型相比，这种模型提供了更好的资源管理：

使用较少的线程便可以处理许多连接，因此也减少了内存管理和上下文切换所带来的开销；
当没有I/O操作需要处理的时候，线程也可以被用于其他任务

Netty

异步与事件驱动

本质上，一个即是异步的又是事件驱动的系统会表现出一种特殊的、对我们来说极具价值的行为：它可以以任意的顺序响应在任意的时间点产生的事件。

这种能力对于实现最高级别的可伸缩性至关重要，定义为：“一种系统、网络或者进程在需要处理的工作不断增长时，可以通过某种可行的方式或扩大它的处理能力适应这种增长的能力。”

异步和可伸缩性之间的联系：

非阻塞网络调用使得我们可以不必等待一个操作的完成。完全异步的I/O正是基于这个特性构建的，并且更进一步：异步方法会立即返回，并且在它完成时，会直接或者在稍后的某个时间点通知用户；
选择器使得我们能够通过较少的线程便可监视许多连接上的事件；

Netty的核心组件

Channel

Channel是Java NIO的一个基本构造，它代表一个到实体（如一个硬件设备、一个文件、一个网络套接字或者一个能够执行一个或者多个不同的I/O操作的程序组件）的开放连接，如读操作和写操作。可以把Channel看作是传入（入站）或者传出（出站）数据的载体。因此，它可以被打开或者被关闭，连接或者断开连接。

回调

一个回调其实就是一个方法，一个指向已经被提供给另外一个方法的方法的引用。这使得后者可以在适当的时候调用前者。

Future

Future提供了另一种在操作完成时通知应用程序的方式。这个对象可以看作是一个异步操作的结果的占位符；它将在未来的某个时刻完成，并提供对其结果的访问。

JDK预置了interface java.util.concurrent.Future，但是其所提供的实现，只允许手动检查对应的操作是否已经完成，或者一直阻塞知道它完成。这是非常繁琐的，所以Netty提供了它自己的实现——ChannelFuture，用于执行异步操作的时候使用。

ChannelFuture提供了几种额外的方法，这些方法使得我们能够注册一个或者多个ChannelFutureListener实例。监听器的回调方法OperationComplete()，将会在对应的操作完成时被调用。然后监听器可以判断该操作是成功地完成了还是出错了。如果是后者，我们可以检索产生的Throwable。简而言之，由ChannelFutureListener提供的通知机制消除了手动检查对应的操作是否完成的必要。

每个Netty的出站I/O操作都将返回一个ChannelFuture；也就是说，它们都不会阻塞，完全是异步和事件驱动的。

private static void nettyTest(){
    Channel channel = new Channel(){};
    // Does not block
    ChannelFuture future = channel.connect(new InetSocketAddress("192.168.0.1", 25)); //异步连接到远程节点
    future.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture channelFuture) throws Exception {
            if(future.isSuccess()){
                ByteBuf buffer = Unpooled.copiedBuffer("Hello", Charset.defaultCharset()); // 如果操作是成功的，则创建一个ByteBuf以持有数据
                ChannelFuture wf = future.channel().writeAndFlush(buffer);  // 将数据异步地发送到远程节点，返回一个ChannelFuture
                //...
            }else{
                Throwable cause = future.cause(); //如果发生错误，则访问描述原因的Throwable
                cause.printStackTrace();
            }
        }
    });
}

事件和ChannelHandler

Netty使用不同的事件来通知我们状态的改变或者是操作的状态。这些动作可能是：

记录日志；
数据转换；
流控制；
应用程序逻辑。

Netty是一个网络编程框架，所以事件是按照它们与入站或出站数据流的相关性进行分类的。可能由入站数据或者相关的状态更改而触发的事件包括：

连接已被激活或者连接失活；
数据读取；
用户事件；
错误事件。

出站事件是未来将会触发的某个动作的操作结果，包括：

打开或关闭到远程节点的连接；
将数据写到或者冲刷到套接字。

Netty应用程序初探

所有的Netty服务器都需要以下两部分：

至少一个ChannelHandler——该组件实现了服务器对从客户端接收的数据的处理，即它的业务逻辑；
引导——这是配置服务器的启动代码。至少，它会将服务器绑定到它要监听连接请求的端口上。

编写Echo服务器

ChannelHandler与业务逻辑

@ChannelHandler.Sharable // 标示一个ChannelHandler可以被多个Channel安全地共享
public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //super.channelRead(ctx, msg);
        ByteBuf in = (ByteBuf)msg;
        System.out.println("Server received: " + in.toString(CharsetUtil.UTF_8)); // 将消息记录到控制台
        ctx.write(in);  // 将接收到的消息写给发送者，而不冲刷出站消息
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //super.channelReadComplete(ctx);
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
                .addListener(ChannelFutureListener.CLOSE); // 将未决消息冲刷到远程节点，并且关闭该Channel
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //super.exceptionCaught(ctx, cause);
        cause.printStackTrace();  // 打印异常栈跟踪
        ctx.close();  // 关闭该Channel
    }
}

如果不捕获异常，会发生什么呢？
每个Channel都拥有一个与之相关联的ChannelPipeline，其持有一个ChannelHandler的实例链。在默认的情况下，ChannelHandler会把对它的方法的调用转发给链中的下一个ChannelHandler。因此，如果exceptionCaught()方法没有被该链中的某处实现，那么所接收的异常将会被传递到ChannelPipeline的尾端并被记录。为此，你的应用程序应该提供至少一个实现了exceptionCaught()方法的ChannelHandler。

关键点：

针对不同类型的事件来调用ChannelHandler；
应用程序通过实现或者扩展ChannelHandler来挂钩到事件的生命周期，并且提供自定义的应用程序逻辑；
在架构上，ChannelHandler有助于保持业务逻辑与网络处理代码的分离。这简化了开发过程，因为代码必须不断地演化以响应不断变化的需求。

引导服务器

package com.louris.springboot.utils;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.net.InetSocketAddress;

public class EchoServer {

    private final int port;

    public EchoServer(int port){
        this.port = port;
    }

    public void start() throws Exception{
        final EchoServerHandler serverHandler = new EchoServerHandler();
        EventLoopGroup group = new NioEventLoopGroup(); // 创建EventLoopGroup
        try{
            ServerBootstrap serverBootstrap = new ServerBootstrap(); // 创建ServerBootstrap
            serverBootstrap.group(group)
                    .channel(NioServerSocketChannel.class)  // 指定所使用的NIO传输Channel
                    .localAddress(new InetSocketAddress(port))  //使用指定的端口设置套接字地址
                    .childHandler(new ChannelInitializer<SocketChannel>() { // 添加一个EchoServerHandler到子Channel的ChannelPipieline
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(serverHandler);  // EchoServerHandler被标注为@Shareable，所以我们可以总是使用同样的实例
                        }
                    });
            ChannelFuture future = serverBootstrap.bind().sync(); // 异步地绑定服务器；调用sync()阻塞等待直到绑定完成
            future.channel().closeFuture().sync(); // 获取Channel的CloseFuture，并且阻塞当前线程直到它完成
        }finally {
            group.shutdownGracefully().sync();
        }
    }

    public static void main(String[] args) throws Exception{
        if(args.length != 1){
            System.err.println("Usage: " + EchoServer.class.getSimpleName() + " <port>");
        }
        int port = Integer.parseInt(args[0]);  // 设置端口值（如果端口参数的格式不正确，则抛出一个NumberFormatException）
        new EchoServer(port).start();  // 调用服务器的start()方法
    }
}

编写Echo客户端

Echo客户端将会：
（1）连接到服务器；
（2）发送一个或者多个消息；
（3）对于每个消息，等待并接收从服务器发回的相同的消息；
（4）关闭连接；

通过ChannelHandler实现客户端逻辑

package com.louris.springboot.netty.client;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;

@ChannelHandler.Sharable  // 标记该类的实例可以被多个Channel共享
public class EchoClientHandler extends  SimpleChannelInboundHandler<ByteBuf>{

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //super.channelActive(ctx);
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!", CharsetUtil.UTF_8));  // 当被通知Channel是活跃的时候，发送一条消息
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
        System.out.println("Client received: " + byteBuf.toString(CharsetUtil.UTF_8));  // 记录已接收消息的转储
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {  // 在发生异常时，记录错误并关闭Channel
        //super.exceptionCaught(ctx, cause);
        cause.printStackTrace();
        ctx.close();
    }
}

每当接收数据时，都会调用channelRead0()方法，需要注意，由服务器发送的消息可能会被分块接收。也就是说，如果服务器发送了5字节，那么不能保证这5字节会被一次性接收。即使是对于这么少量的数据，channelRead0()方法也可能会被调用两次，第一次使用一个持有3字节的ByteBuf(Netty的字节容器)，第二次使用一个持有2字节的ByteBuf。

SimpleChannelInboundHandler与ChannelInboundHandler：
为何客户端与服务端使用不同？这和两个因素的相互作用有关；业务逻辑如何处理消息以及Netty如何管理资源：

在客户端，当channelRead0()方法完成时，你已经有了传入消息，并且已经处理完它了。当该方法返回时，SimpleChannelInBoundHandler负责释放指向保存该消息的ByteBuf的内存引用。
在服务端EchoServerHandler汇总，你仍然需要将传入消息会送给发送者，而write()操作是异步的，直到channelRead()方法返回后可能仍然没有完成。为此，EchoServerHandler扩展了ChannelInboundHandlerAdapter，其在这个时间点上不会释放消息。
消息在EchoServerHandler的channelReadComplete()方法中，当writeAndFlush()方法被调用时被释放。

引导客户端

package com.louris.springboot.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.net.InetSocketAddress;

public class EchoClient {

    private final String host;
    private final int port;

    public EchoClient(String host, int port){
        this.host = host;
        this.port = port;
    }

    public void start() throws Exception{
        EventLoopGroup group = new NioEventLoopGroup();
        try{
            Bootstrap bootstrap = new Bootstrap(); // 创建Bootstrap
            bootstrap.group(group)  // 指定EventLoopGroup以处理客户端事件；需要适用于NIO的实现
                    .channel(NioServerSocketChannel.class)  // 适用于NIO传输的Channel类型
                    .remoteAddress(new InetSocketAddress(host, port))   // 设置服务器的InetSocketAddress
                    .handler(new ChannelInitializer<SocketChannel>(){  // 在创建Channel时，向ChannelPipeline中添加一个EchoClientHandler实例
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new EchoClientHandler());
                        }
                    });
            ChannelFuture future = bootstrap.connect().sync(); // 连接到远程节点，阻塞等待直到连接完成
            future.channel().closeFuture().sync();  // 阻塞，直到Channel关闭
        }finally {
            group.shutdownGracefully().sync();  // 关闭线程池并且释放所有的资源
        }
    }

    public static void main(String[] args) throws Exception{
        if(args.length != 2){
            System.err.println("Usage: " + EchoClient.class.getSimpleName() + " <host> <port>");
            return;
        }
        String host = args[0];
        int port = Integer.parseInt(args[1]);
        new EchoClient(host, port).start();
    }
}

SpringBoot启动测试

服务器端

package com.louris.springboot.configs;

import com.louris.springboot.utils.EchoServer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ServerConfig {

    @Bean
    EchoServer createEchoServer(){
        return new EchoServer(10000);
    }
}

package com.louris.springboot;

import com.louris.springboot.utils.EchoServer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;

@SpringBootApplication
public class ServerApplication {

    //@Autowired
    //private static EchoServer echoServer;

    public static void main(String[] args) throws Exception{

        ApplicationContext applicationContext = SpringApplication.run(ServerApplication.class, args);
        EchoServer echoServer = applicationContext.getBean(EchoServer.class);
        echoServer.start();
    }

}

客户端

package com.louris.springboot.configs;

import com.louris.springboot.utils.EchoClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ClientConfig {

    @Bean
    EchoClient createClient(){
        return new EchoClient("localhost", 10000);
    }
}

package com.louris.springboot;

import com.louris.springboot.utils.EchoClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;

@SpringBootApplication
public class ClientApplication {


    public static void main(String[] args) throws Exception{

        ApplicationContext applicationContext = SpringApplication.run(ClientApplication.class, args);
        EchoClient echoClient = applicationContext.getBean(EchoClient.class);
        echoClient.start();
    }

}

Netty的组件和设计

Channel、EventLoop和ChannelFuture

Channel——Socket
EventLoop——控制流、多线程处理、并发
ChannelFuture——异步通知

Channel接口

基本的I/O操作（bind()、connect()、read()和write()）依赖于底层网络传输所提供的原语。在基于Java的网络编程中，其基本的构造是class Socket。Netty的Channel接口所提供的API，大大降低了直接使用Socket类的复杂性。此外，Channel也是拥有许多预定义、专门化实现的广泛类层次结构的根，下面是一个简短的部分清单：

EmbeddedChannel
LocalServerChannel
NioDatagramChannel
NioSctpChannel
NioSocketChannel

EventLoop接口

EventLoop定义了Netty的核心抽象，用于处理连接的生命周期中所发生的事件。关系如图所示：

一个EventLoopGroup包含一个或多个EventLoop
一个EventLoop在它的生命周期内只和一个Thread绑定
所有由EventLoop处理的I/O事件都将在它专有的Thread上被处理
一个Channel在它的生命周期内只注册一个EventLoop
一个EventLoop可能会被分配给一个或多个Channel

注意，在这种设计中，一个给定Channel的I/O操作都是由相同的Thread执行的，实际上消除了对于同步的需要。

ChannelFuture接口

Netty中所有的I/O操作都是异步的。因为一个操作可能不会立即返回，所以需要一种用于在之后的某个时间点确定其结果的方法。为此，Netty提供了ChannelFuture接口，其addListener()方法注册了一个ChannelFutureListener，以便在某个操作完成时（无论是否成功）得到通知。

ChannelHandler和ChannelPipeline

ChannelHandler接口

从应用程序开发人员的角度来看，Netty的主要组件是ChannelHandler，它充当了所有处理入站和出站数据的应用程序逻辑的容器。这是可行的，因为ChannelHandler的方法是由网络事件触发的。事实上，ChannelHandler可专门用于几乎任何类型的动作，例如将数据从一种格式转换为另外一种格式，或者处理转换过程中所抛出的异常。

ChannelPipeline接口

ChannelPipeline提供了ChannelHandler链的容器，并定义了用于在该链上传播入站和出站事件流的API。当Channel被创建时，它会被自动地分配到它专属的ChannelPipeline。

ChannelHandler类的层次结构：

ChannelHandler是专为支持广泛的用于而设计的，可以将它看作是处理往来ChannelPipeline事件（包括数据）的任何代码的通用容器。使得事件刘静ChannelPipeline是ChannelHandler的工作，它们是在应用程序的初始化或者引导阶段被安装的。这些对象接收事件、执行它们所实现的处理逻辑，并将数据传递给链中的下一个ChannelHandler。它们的执行顺序是由它们被添加的顺序所决定的。实际上，而被称为ChannelPipeline的是这些ChannelHandler的编排顺序。

Netty应用程序中入站和出站数据流之间的区别：

当ChannelHandler被添加到ChannelPipeline时，它将会被分配到一个ChannelHandlerContext，其代表了ChannelHandler和ChannelPipeline之间的绑定。虽然这个对象可以被用于获取底层的Channel，但是它主要还是被用于写出站数据。

在Netty中，有两种发送消息的方式：

直接写到Channel中：导致消息从ChannelPipeline的尾端开始流动；
写到和ChannelHandler相关联的ChannelHandlerContext对象中：导致消息从ChannelPipeline中的下一个ChannelHandler开始流动。

ChannelHandlerAdapter

Netty以适配类的形式提供了大量默认的ChannelHandler实现，其旨在简化应用程序处理逻辑的开发过程。这些适配器类及它们的子类将自动执行这个操作，所以可以只重写那些想要特殊处理的方法和事件。

常见适配器类：

ChannelHandlerAdapter
ChannelInboundHandlerAdapter
ChannelOutboundHandlerAdapter
ChannelDuplexHandler

编码器和解码器

当通过Netty发送或者接收一个消息的时候，就将会发生一次数据转换。入站消息会被解码；也就是说，从字节转换为另一种格式，通常是一个Java对象。如果是出站消息，则会发生相反方向的转换：它将从它的当前格式被编码为字节。这两种方向的转换的原因很简单：网络数据总是一系列的字节。

这些基类的名称类似于ByteToMessageDecoder或MessageToByteEncoder。
所有由Netty提供的编码器/解码器适配器类都实现了ChannelOutboundHandler或者ChannelInboundHandler接口。
对于入站数据来说，channelRead方法/事件已经被重写了。对于每个从入站Channel读取的消息，这个方法都将会被调用。随后，它将调用由预置解码器所提供的decode()方法，并将已解码的字节转发给ChannelPipeline中的下一个ChannelInboundHandler。
出栈消息的模式是相反方向：编码器将消息转换为字节，并将它们转发给下一个ChannelOutboundHandler。

抽象类SimpleChannelInboundHandler

最常见的情况是利用一个ChannelHandler来接收解码消息，并对该数据应用业务逻辑。要创建一个这样的ChannelHandler，只需要扩展基类SimpleChannelInboundHandler，其中T是要处理的消息的Java类型。在这个ChannelHandler中，将需要重写基类的一个或者多个方法，并且获取一个到ChannelHandlerContext的引用，这个引用将作为参数传递给ChannelHandler的所有方法。

在这种类型的ChannelHandler中，最重要的方法是channelRead0(ChannelHandlerContext, T)。除了要求不要阻塞当前的I/O线程之外，其具体实现完全取决于用户。

引导

Netty的引导类为应用程序的网络层配置提供了容器，这涉及将一个进程绑定到某个指定的端口，或者将一个进程连接到另一个运行在某个指定主机的指定端口上的进程。

两种类型的引导：

一种用于客户端：Bootstrap
一种用于服务器：ServerBootstrap

类别	Bootstrap	ServerBootstrap
网络编程中的作用	连接到远程主机和端口	绑定到一个本地端口
EventLoopGroup的数目	1	2

服务器需要2个EventLoopGroup的原因是服务器需要两组不同的Channel：

第一组将只包含一个ServerChannel，代表服务器自身的已绑定到某各本地端口的正在监听的套接字；
第二组将包含所有已创建的用来处理传入客户端连接（对于每个服务器已经接受的连接都有一个）的Channel。

与ServerChannel相关联的EventLoopGroup将分配一个负责为传入连接请求创建Channel的EventLoop。一旦连接被接受，第二个EventLoopGroup就会给它的Channel分配一个EventLoop。

传输

案例研究：传输迁移

传统JDK API的BIO和NIO

BIO

package com.louris.springboot.utils;

import java.io.IOException;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.Charset;

public class PlainBioServer {
    public void serve(int port) throws IOException{
        final ServerSocket socket = new ServerSocket(port);  // 将服务器绑定到指定端口
        try{
            for(;;){
                final Socket clientSocket = socket.accept();  // 接受连接
                System.out.println("Accepted connection from " + clientSocket);
                new Thread(new Runnable() {  // 创建一个新的线程来处理该连接
                    @Override
                    public void run() {
                        OutputStream os;
                        try{
                            os = clientSocket.getOutputStream();
                            os.write("Hi!\r\n".getBytes(Charset.forName("UTF-8")));  // 将消息写给已连接的客户端
                        }catch (IOException e){
                            e.printStackTrace();
                        }finally {
                            try{
                                clientSocket.close();  // 关闭连接
                            }catch (IOException e){
                                e.printStackTrace();
                                // ignore on close
                            }
                        }
                    }
                }).start();  // 启动线程
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }
}

BIO可以处理中等数量的并发客户端。但是随着应用程序变得流行起来，会发现它并不能很好地伸缩到支撑成千上万的并发连入连接。

NIO(基于Selector的IO多路复用)

package com.louris.springboot.utils;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

public class PlainNioServer {
    public void serve(int port) throws IOException{
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        ServerSocket serverSocket = serverSocketChannel.socket();
        InetSocketAddress address = new InetSocketAddress(port);
        serverSocket.bind(address);  // 将服务器绑定到选定的端口
        Selector selector = Selector.open();  // 打开Selector来处理Channel
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);  // 将ServerSocket注册到Selector以接受连接
        final ByteBuffer msg = ByteBuffer.wrap("Hi!\r\n".getBytes());
        for(;;){
            try{    // 等待需要处理的新事件；阻塞将一直持续到下一个传入事件
                selector.select();
            }catch (IOException e){
                e.printStackTrace();
                // handle exception
                break;
            }
            Set<SelectionKey> readyKeys = selector.selectedKeys();  // 获取所有接收事件的SelectionKey实例
            Iterator<SelectionKey> iterator = readyKeys.iterator();
            while(iterator.hasNext()){
                SelectionKey key = iterator.next();
                iterator.remove();
                try{
                    if(key.isAcceptable()){  // 检查事件是否是一个新的已经就绪可以被接受的连接
                        ServerSocketChannel server = (ServerSocketChannel) key.channel();
                        SocketChannel client = server.accept();
                        client.configureBlocking(false);
                        client.register(selector, SelectionKey.OP_WRITE | SelectionKey.OP_READ, msg.duplicate());  // 接受客户端，并将它注册到选择器
                        System.out.println("Accepted connection from " + client);
                    }
                    if(key.isWritable()){  // 检查套接字是否已经准备好写数据
                        SocketChannel client = (SocketChannel) key.channel();
                        ByteBuffer buffer = (ByteBuffer)key.attachment();
                        while(buffer.hasRemaining()){
                            if(client.write(buffer) == 0) break;  // 将数据写到已连接的客户端
                        }
                        client.close();  // 关闭连接
                    }
                }catch (IOException e){
                    key.cancel();
                    try{
                        key.channel().close();
                    }catch (IOException ioe){
                        //ignore on close
                    }
                }
            }
        }
    }
}

通过Netty使用BIO和NIO

BIO

package com.louris.springboot.utils;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.oio.OioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.oio.OioServerSocketChannel;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;

public class NettyBioServer {
    public void server(int port) throws Exception {
        final ByteBuf buf = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi\r\n", Charset.forName("UTF-8")));
        EventLoopGroup group = new OioEventLoopGroup();
        try{
            ServerBootstrap serverBootstrap = new ServerBootstrap();  // 创建ServerBootstrap
            serverBootstrap.group(group)
                    .channel(OioServerSocketChannel.class)  // 使用OioEventLoopGroup以允许阻塞模式（旧的I/O）
                    .localAddress(new InetSocketAddress(port))
                    .childHandler(new ChannelInitializer<SocketChannel>() {  // 指定ChannelInitializer，对于每个已接受的连接都调用它
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){ // 添加一个ChannelInboundHandlerAdapter以拦截和处理事件
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    //super.channelActive(ctx);
                                    ctx.writeAndFlush(buf.duplicate())
                                            .addListener(ChannelFutureListener.CLOSE);  // 将消息写到客户端，并添加ChannelFutureListener，以便消息一被写完就关闭连接
                                }
                            });
                        }
                    });
            ChannelFuture future = serverBootstrap.bind().sync();  // 绑定服务器以接受连接
            future.channel().closeFuture().sync();
        }finally {
            group.shutdownGracefully().sync(); // 释放所有的资源
        }
    }
}

NIO

package com.louris.springboot.utils;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
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.net.InetSocketAddress;
import java.nio.charset.Charset;

public class NettyNioServer {
    public void server(int port) throws Exception {
        final ByteBuf buf = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi\r\n", Charset.forName("UTF-8")));
        EventLoopGroup group = new NioEventLoopGroup(); // 为非阻塞模式使用NioEventLoopGroup
        try{
            ServerBootstrap serverBootstrap = new ServerBootstrap();  // 创建ServerBootstrap
            serverBootstrap.group(group)
                    .channel(NioServerSocketChannel.class)  // 使用OioEventLoopGroup以允许阻塞模式（旧的I/O）
                    .localAddress(new InetSocketAddress(port))
                    .childHandler(new ChannelInitializer<SocketChannel>() {  // 指定ChannelInitializer，对于每个已接受的连接都调用它
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){ // 添加一个ChannelInboundHandlerAdapter以拦截和处理事件
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    //super.channelActive(ctx);
                                    ctx.writeAndFlush(buf.duplicate())
                                            .addListener(ChannelFutureListener.CLOSE);  // 将消息写到客户端，并添加ChannelFutureListener，以便消息一被写完就关闭连接
                                }
                            });
                        }
                    });
            ChannelFuture future = serverBootstrap.bind().sync();  // 绑定服务器以接受连接
            future.channel().closeFuture().sync();
        }finally {
            group.shutdownGracefully().sync(); // 释放所有的资源
        }
    }
}

可以看到两者的API相同，改动很小，无论选用哪一种传输的实现，代码都仍然几乎不受影响。在所有情况下，传输的实现都依赖于interface Channel、ChannelPipeline和ChannelHandler。

传输API

传输API的核心是interface Channel，它被用于所有的I/O操作。Channel类的层次结构如下：

每个Channel都将会被分配一个ChannelPipeline和ChannelConfig；
ChannelConfig包含了该Channel的所有配置设置，并且支持热更新。由于特定的传输可能具有独特的设置，所以它可能会实现一个ChnanelConfig的子类型；
由于Channel是独一无二的，所以为了保证顺序将Channel声明为java.lang.Comparable的一个子接口。因此，如果两个不同的Channel实例都返回了相同的散列码，那么AbstractChannel中的compareTo()方法的实现将会抛出一个Error；
ChannelPipeline持有所有将应用于入站和出站数据以及事件的ChannelHandler实例，这些ChannelHandler实现了应用程序用于处理状态变化以及数据处理的逻辑。

ChannelHandler的典型用途包括：

将数据从一种格式转换为另一种格式；
提供异常的通知；
提供Channel变为活动的或者非活动的通知；
提供当Channel注册到EventLoop或者从EventLoop注销时的通知；
提供有关用户自定义事件的通知

拦截过滤器：ChannelPipeline实现了一种常见的设计模式——拦截过滤器（Intercepting Filter）。UNIX管道是另一个熟悉的例子：多个命令被链接在一起，其中一个命令的输出端将连接到命令行中下一个命令的输入端。

除了访问所分配的ChannelPipeline和ChannelConfig之外，也可以利用Channel的其他方法：

方法名	描述
eventLoop	返回分配给Channel的EventLoop
pipeline	返回分配给Channel的ChannelPipeline
isActive	如果Channel是活动的，则返回true。活动的意义可能依赖于底层的传输。例如一个Socket传输一旦连接到远程节点便是活动的，而一个Datagram传输一旦被打开便是活动的
localAddress	返回本地的SocketAddress
remoteAddress	返回远程的SocetAddress
write	将数据写到远程节点。这个数据将被传递给ChannelPipeline，并且排队直到它被冲刷
flush	将之前已写的数据冲刷到底层传输，如一个Socket
writeAndFlush	一个简便的方法，等同于调用write()并接着调用flush()

写数据并冲刷到远程节点

Channel channel = ...
ByteBuf buf = Unpooled.copiedBuffer("your data", CharsetUtil.UTF_8);  // 创建持有要写数据的ByteBuf
ChannelFuture channelFuture = channel.writeAndFlush(buf); // 写数据并冲刷它
channelFuture.addListener(new ChannelFutureListener() { // 添加ChannelFutureListener以便在写操作完成后接收通知
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        if(future.isSuccess()){ // 写操作完成，并且没有错误发生
            System.out.println("Write successful");
        }else{
            System.err.println("Write error"); // 记录错误
            future.cause().printStackTrace();
        }
    }
});

Netty的Channel实现是线程安全的，因此可以存储一个到Channel的引用，并且每当需要向远程节点写数据时，都可以使用它，即使当时许多线程都在使用它。多线程写数据的例子如下，需要注意消息将会被保证按顺序发送。

final Channel channel = ...
final ByteBuf buf = Unpooled.copiedBuffer("your data", CharsetUtil.UTF_8).retain(); // 创建持有要写数据的ByteBuf
Runnable writer = new Runnable() { // 创建将数据写到Channel的Runnable
    @Override
    public void run() {
        channel.writeAndFlush(buf.duplicate());
    }
};

Executor executor = Executors.newCachedThreadPool();  // 获取到线程池Executor的引用

// write in one thread
executor.execute(writer); // 递交写任务给线程池以便在某各线程中执行

// write in another thread
executor.execute(writer); // 递交另一个写任务以便在另一个线程中执行

...

内置的传输

Netty内置了一些可开箱即用的传输。因为并不是它们所有的传输都支持每一种协议，所以必须选择一个和应用程序所使用的协议相容的传输。

名称	包	描述
NIO	io.netty.channel.socket.nio	使用java.nio.channels包作为基础——基于选择器的方式
Epoll	io.netty.channel.epoll	由JNI驱动的epoll()和非阻塞IO。这个传输支持只有在Linux上可用的多种特性，如SO_RESEPORT，比NIO传输更快，而且是完全非阻塞的
OIO	io.netty.channel.socket.ioi	使用java.net包作为基础——使用阻塞流
Local	io.netty.channel.local	可以在VM内部通过管道进行通信的

NIO——非阻塞I/O

NIO提供了一个所有I/O操作的全异步的实现。它利用了自NIO子系统被引入JDK1.4时便可用的基于选择器的API。

选择器背后的基本概念是充当一个注册表，在那里将可以请求在Channel的状态发生变化时得到通知。可能的状态变化有：

新的Channel已被接受并且就绪；
Channel连接已经完成；
Channel有已经就绪的可供读取的数据；
Channel可用于写数据。

选择器运行在一个检查状态变化并对其做出响应的线程上，在应用程序对状态的改变做出响应之后，选择器将会被重置，并将重复这个过程。

下表代表了由class java.nio.channels.SelectionKey定义的位模式。这些位模式可以组合起来定义一组应用程序正在请求通知的状态变化集。

名称	描述
OP_ACCEPT	请求在接受新连接并创建Channel时获得通知
OP_CONNECT	请求在建立一个连接时获得通知
OP_READ	请求当数据已经就绪，可以从Channel中读取时获取通知
OP_WRITE	请求当可以向Channel中写更多的数据时获得通知。这处理了套接字缓冲区被完全填满时的情况，这种情况通常发生在数据的发送速度比远程节点可处理的速度更快的时候

非阻塞I/O流程图：

Epoll——用于Linux的本地非阻塞传输

Netty的NIO传输基于Java提供的异步/非阻塞网络编程的通用抽象。虽然这保证了Netty的非阻塞API可以在任何平台上使用，但它也包含了相应的限制，因为JDK为了在所有系统上提供相同的功能，必须做出妥协。

Linux作为高性能网络编程的平台，其重要性与日俱增，这催生了大量先进特性的开发，其中包括epoll——一个高度可扩展的I/O事件通知特性。这个API自Linux内核版本2.5.44(2022)被引入，提供了比旧的POSIX selector和poll系统调用更好的性能，同时现在也是Linux上非阻塞网络编程的事实标准。Linux JDK NIO API使用了这些epoll调用。

Netty为Linux提供了一组NIO API，其以一种和它本身的设计更加一致的方式使用epoll，并且以一种更加轻量的方式使用中断。高负载下，它的性能要优于JDK的NIO实现。

JDK的实现是水平触发，而Netty的（默认的）是边沿触发。

package com.louris.springboot.utils;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.socket.SocketChannel;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;

public class NettyNioServer {
    public void server(int port) throws Exception {
        final ByteBuf buf = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi\r\n", Charset.forName("UTF-8")));
        EventLoopGroup group = new EpollEventLoopGroup(); //new NioEventLoopGroup(); // 为非阻塞模式使用NioEventLoopGroup
        try{
            ServerBootstrap serverBootstrap = new ServerBootstrap();  // 创建ServerBootstrap
            serverBootstrap.group(group)
                    .channel(EpollServerSocketChannel.class)  // 使用OioEventLoopGroup以允许阻塞模式（旧的I/O）
                    .localAddress(new InetSocketAddress(port))
                    .childHandler(new ChannelInitializer<SocketChannel>() {  // 指定ChannelInitializer，对于每个已接受的连接都调用它
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){ // 添加一个ChannelInboundHandlerAdapter以拦截和处理事件
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    //super.channelActive(ctx);
                                    ctx.writeAndFlush(buf.duplicate())
                                            .addListener(ChannelFutureListener.CLOSE);  // 将消息写到客户端，并添加ChannelFutureListener，以便消息一被写完就关闭连接
                                }
                            });
                        }
                    });
            ChannelFuture future = serverBootstrap.bind().sync();  // 绑定服务器以接受连接
            future.channel().closeFuture().sync();
        }finally {
            group.shutdownGracefully().sync(); // 释放所有的资源
        }
    }
}

OIO——旧的阻塞I/O

Netty的OIO传输实现代表了一种折中：它可以通过常规的传输API使用，但是由于它是建立在java.net包的阻塞实现之上的，所以它不是异步的。但是，它仍然非常适合于某些用途。

在Java.net API中，我们通常会有一个用于接受到达正在监听的ServerSocket的新连接的线程。会创建一个新的和远程节点进行交互的套接字，并且分配一个新的用于处理相应通信流量的线程。这是必需的，因为某个指定套接字上的任何I/O操作在任意的时间点上都可能会阻塞。使用单个线程来处理多个套接字，很容易导致一个套接字上的阻塞操作也捆绑了所有其他的套接字。

Netty不同，Netty利用了SO_TIMEOUT这个Socket标志，它指定了等待一个I/O操作完成的最大毫秒数。如果操作在指定的时间间隔内没有完成，则将会抛出一个SocketTimeoutException。Netty将捕获这个异常并继续处理循环。在EventLoop下一次运行时，它将再次尝试。这实际上也是类似于Netty这样的异步框架能够支持OIO的唯一方式。

OIO流程图：

用于JVM内部通信的Local传输

Netty提供了一个Local传输，用于在同一个JVM中运行的客户端和服务器之间的异步通信。同样，这个传输也支持所有Netty传输实现都共同的API。

在这个传输中，和服务器Channel相关联的SocketAddress并没有绑定物理网络地址；相反，只要服务器还在运行，它就会被存储在注册表里，并在Channel关闭时注销。因为这个传输并不接受真正的网络流量，所以它并不能够和其他传输实现进行互操作。因此，客户端希望连接到（在同一个JVM中）使用了这个传输的服务器端时也必须使用它。除了这个限制，它的使用方式和其他传输一模一样。

Embedded传输

Netty提供了一种额外的传输，使得你可以将一组ChannelHandler作为帮助器类嵌入到其他的ChannelHandler内部。通过这种方式，你将可以扩展一个ChannelHandler的功能，而又不需要修改其内部代码。其关键是一个EmbeddedChannel的具体的Channel的实现。

传输的用例

传输	TCP	UDP	SCTP	UDT
NIO	×	×	×	×
Epoll（仅Linux）	×	×	-	-
OIO	×	×	×	×

应用程序的需求	推荐的传输
非阻塞代码库或者一个常规的起点	NIO（或者在Linux上使用epoll）
阻塞代码库	OIO
在同一个JVM内部的通信	Local
测试ChannelHandler的实现	Embedded

ByteBuf

Java NIO提供了ByteBuffer作为它的字节容器，但是这个类使用起来过于复杂，而且也有些繁琐。

Netty的数据容器为ByteBuf，一个强大的实现，既解决了JDK API的局限性，又为网络应用程序的开发者提供了更好的API。

ByteBuf的API

abstract class ByteBuf
interface ByteBufHolder

ByteBuf API的优点：

它可以被用户自定义的缓冲区类型扩展；
通过内置的符合缓冲区类型实现了透明的零拷贝；
容量可以按需增长（类似于JDK的StringBuilder）；
在读和写这两种模式之间切换不需要调用ByteBuffer的flip()方法；
读和写使用了不同的索引；
支持方法的链式调用；
支持引用计数；
支持池化；

ByteBuf类

如何工作

ByteBuf维护了两个不同的索引：一个用于读取，一个用于写入。

readerIndex
writerIndex

ByteBuf的使用模式

堆缓冲区

最常用的ByteBuf模式是将数据存储在JVM的堆空间中。这种模式被称为支撑数组（backing array），它能在没有使用池化的情况下提供快速的分配和释放。这种方式，非常适合于有遗留的数据需要处理的情况。

ByteBuf heapBuf = Unpooled.buffer(100); // 堆缓冲区
if(heapBuf.hasArray()){  // 检查ByteBuf是否有一个支撑数组
    byte[] array = heapBuf.array();  // 如果有，则获取对该数组的引用
    int offset = heapBuf.arrayOffset() + heapBuf.readerIndex();  // 计算第一个字节的偏移量
    int length = heapBuf.readableBytes();
    handleArray(array, offset, length);
}

直接缓冲区

直接缓冲区使用的是物理内存，避免在每次调用本地I/O操作之前（或者之后）将缓冲区的内容复制到一个中间缓冲区（或者从中间缓冲区把内容复制到缓冲区）；
直接缓冲区的内容将驻留在常规的会被垃圾回收的堆之外；
直接缓冲区的主要缺点是，相对于基于堆的缓冲区，它们的分配和释放都较为昂贵；另外如果处理遗留代码，因为数据不在堆上，不得不进行一次复制。

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

复合缓冲区

复合缓冲区为多个ByteBuf提供一个聚合视图，在这里可以根据需要添加或者删除ByteBuf实例；
Netty通过一个ByteBuf子类——CompositeByteBuf——实现了这个模式，它提供了一个将多个缓冲区表示为单个合并缓冲区的虚拟表示；
CompositeByteBuf中的ByteBuf实例可能同时包含直接内存分配和非直接内存分配：如果其中只有一个实例，那么对CompositeByteBuf上的hasArray()方法的调用将返回该组件上的hasArray()方法的值；否则它将返回false；
可以使得多个消息重用相同的消息主题，不用每个消息都重新分配缓冲区；
Netty使用CompositeByteBuf来优化套接字的I/O操作，尽可能地消除了由JDK的缓冲区实现所导致的性能以及内存使用率的惩罚。

CompositeByteBuf messageBuf = Unpooled.compositeBuffer();
ByteBuf headerBuf = Unpooled.buffer(100); // can be backing or direct
ByteBuf bodyBuf = Unpooled.directBuffer(100); // can be backing or direct
messageBuf.addComponents(headerBuf, bodyBuf);  // 将ByteBuf实例追加到CompositeByteBuf

// do something

messageBuf.removeComponent(0); // remove the header  // 删除位于索引位置为0（第一个组件）的ByteBuf
for(ByteBuf buf:messageBuf){
    System.out.println(buf.toString());
}

// 类似于访问直接缓冲区的模式
int length = messageBuf.readableBytes(); // 获得可读字节数
byte[] array = new byte[length]; // 分配一个具有可读字节数长度的新数组 
messageBuf.getBytes(messageBuf.readerIndex(), array); //将字节读到该数组中
handleArray(array, 0, array.length);

字节级操作

随机访问索引

ByteBuf buffer = Unpooled.buffer(); // Unpooled.directBuffer();
for(int i = 0; i < buffer.capacity(); i ++){
    byte b = buffer.getByte(i);
    System.out.println((char)b);
}

顺序访问索引

可丢弃字节

可丢弃字节的分段包含了已经被读过的字节。通过调用discardReadBytes()方法，可以丢弃它们并回收空间。这个分段的初试大小为0，存储在readerIndex中，会随着read操作的执行而增加(get方法不会移动readerIndex)。

readerIndex(index)或writerIndex(index)可以手动移动索引。

丢弃已读字节之后的ByteBuf:

可读字节

ByteBuf的可读字节分段存储了实际数据。新分配的、包装的或者复制的缓冲区的默认的readerIndex值为0。任何名称以read或者skip开头的操作都将检索或者跳过位于当前readerIndex的数据，并且将它增加已读字节数。
如果被调用的方法需要一个ByteBuf参数作为写入的目标，并且没有指定目标索引参数，那么该目标缓冲区的writerIndex也将被增加，例如readBytes(ByteBuf des)；
如果尝试在缓冲区的可读字节数已经耗尽时从中读取数据，那么将会引发一个IndexOutOfBoundsException。

ByteBuf buffer = Unpooled.buffer(); // Unpooled.directBuffer();
while(buffer.isReadable()){
    System.out.println(buffer.readByte());
}

可写字节

可写字节分段是指一个拥有未定义内容的、写入就绪的内存区域。新分配的缓冲区的writerIndex的默认值为0；
任何名称以write开头的操作都将从当前的writerIndex处开始写数据，并将它增加已经写入的字节数。如果写操作的目标也是ByteBuf，并且没有指定源索引的值，则源缓冲区的readerIndex也同样会被增加相同的大小；

ByteBuf buffer = Unpooled.buffer(); // Unpooled.directBuffer();

Random random = new Random();
while(buffer.writableBytes() >= 4){
    buffer.writeInt(random.nextInt());
}

索引管理

JDK的InputStream定义了mark(int readlimit)和reset()方法，这些方法分别被用来将流中的当前位置标记为指定的值，以及将流重置到该位置；
同样，可以通过调用markReaderIndex()、markWriterIndex()、resetWriterIndex()和resetReaderIndex()来标记和重置ByteBuf的readerIndex和writerIndex；
通过调用readerIndex(int)或者writerIndex(int)来将索引移动到指定位置；视图将任何一个索引设置到一个无效的位置都将导致一个IndexOutOfBoundsException；
可以通过调用clear()方法来将readerIndex核writerIndex都设置为0.注意，这并不会清除内存中的内容；
调用clear()比调用discardReadBytes()轻量得多，因为它将只是重置索引而不会复制任何的内存。

clear()方法被调用前：

clear()方法被调用后：

查找操作

在ByteBuf中有多重可以用来确定指定值的索引的方法：

最简单的是使用indexOf()方法；
较复杂的查找可以通过那些需要一个ByteBufProcessor作为参数的方法打成；这个接口只定义了一个方法boolean process(byte value);
它将检查输入值是否是正在查找的值

1
2
3

ByteBuf buffer = Unpooled.buffer(); // Unpooled.directBuffer();

int index = buffer.forEachByte(ByteBufProcessor.FIND_CR)

派生缓冲区

派生缓冲区为ByteBuf提供了以专门的方式来呈现其内容的视图。这类视图是通过以下方法被创建的：

duplicate();
slice();
slice(int, int);
UnpooledunmodifiableBuffer(…);
order(ByteOrder);
readSlice(int);

每个这些方法都将返回一个新的ByteBuf实例，它具有自己的读索引、写索引和标记索引，共享的，如果修改了内容，也同时修改了其对应的源实例。

ByteBuf复制：如果需要一个现有缓冲区的真实副本，请使用copy(...)方法。不同于派生缓冲区，由这个调用所返回的ByteBuf拥有独立的数据副本。

对ByteBuf进行切片：

Charset utf8 = Charset.forName("UTF-8");
ByteBuf buf = Unpooled.copiedBuffer("Netty in Action rocks!", utf8);
ByteBuf sliced = buf.slice(0, 15);  // 创建该ByteBuf从索引0开始到索引15结束的一个新切片
System.out.println(sliced.toString(utf8));  // 将打印“Netty in Action"
buf.setByte(0, (byte)'J');  // 更新索引0处的字节
assert buf.getBytes(0) == sliced.getBytes(0); // 将会成功，因为数据是共享的

复制一个ByteBuf:

Charset utf8 = Charset.forName("UTF-8");
ByteBuf buf = Unpooled.copiedBuffer("Netty in Action rocks!", utf8);
ByteBuf copy = buf.copy(0, 15);
System.out.println(sliced.toString(utf8));  // 将打印“Netty in Action"
buf.setByte(0, (byte)'J');  // 更新索引0处的字节
assert buf.getBytes(0) != copy.getBytes(0); // 将会成功，因为数据不是共享的

读/写操作

get()和set操作，从给定索引开始，并且保持索引不变：
read()和write()操作，从给定的索引开始，并且会根据已经访问的字节数对索引进行调整；
getBoolean(int)、getByte(int)、getInt(int)等
setBoolean(int)、setByte(int index, int value)等
readBoolean()、readByte()，readLong()等
writeBoolean(boolean)、writeByte(int)、writeInt(int)等

名称	描述
isReadable()	如果至少有一个字节可供读取，则返回true
isWritable()	如果至少有一个字节可被写入，则返回true
resableBytes()	返回可被读取的字节数
writableBytes()	返回可被写入的字节数
capacity()	返回ByteBuf可容纳的字节数。在此之后，它会尝试再次扩展直到达到maxCapacity()
maxCapacity()	返回ByteBuf可以容纳的最大字节数
hasArray()	如果ByteBuf由一个字节数组支撑，则返回true
array()	如果ByteBuf由一个字节数组支撑则返回该数组；否则，它将抛出一个UnsupportedOperationException异常

ByteBufHolder接口

除了实际的数据负载之外，还需要存储各种属性值。为此，Netty提供了ByteBufHolder，其也为Netty的高级特性提供了支持，如缓冲区池化，其中可以从池中借用ByteBuf，并且在需要自动释放。

名称	描述
content()	返回由这个ByteBufHolder所持有的ByteBuf
copy()	返回这个ByteBufHolder的一个深拷贝，包括一个其所包含的ByteBuf的非共享拷贝
duplicate()	返回这个ByteBufHolder的一个浅拷贝，包括一个其所包含的ByteBuf的共享拷贝

ByteBuf分配

管理ByteBuf实例的不同方式。

按需分配：ByteBufAllocator接口

Channel channel = ...;
ByteBufAllocator allocator = channel.alloc();  // 从Channel获取一个到ByteBufAllocator的引用
...
ChannelHandlerContext ctx = ...;
ByteBufAllocator allocator1 = ctx.alloc();  // 从ChannelHandlerContext获取一个到ByteBufAllocator的引用
...

Netty提供的两种ByteBufAllocator的实现：

PooledByteBufAllocator(默认)：池化了ByteBuf的实例以提高性能并最大限度地减少内存碎片。此实现使用了一种称为jemalloc的已被大量现代操作系统所采用的高效方法来分配内存。
UnpooledByteBufAllocator：实现不池化的ByteBuf实例，并且在每次它被调用时都会返回一个新的实例。

Unpooled缓冲区

可能某些情况下，未能获得一个到ByteBufAllocator的引用。对于这种情况，Netty提供Unpooled的工具类。

具体方法见API

ByteBufUtil类

ByteBufUtil提供了用于操作ByteBuf的静态的辅助方法。因为这个API是通用的，并且和池化无关，所以这些方法已然在分配类的外部实现。
最优价值的可能是hexdump()方法，它以十六进制的表示形式打印ByteBuf的内容。这在各种情况下都很有用，例如，出于调试的目的记录ByteBuf的内容。十六进制的表示通常会提供一个比字节值的直接表示形式更加有用的日志条目，此外，十六进制的版本还可以很容易地转换回实际的字节表示；
另一个有用的方法是boolean equals(ByteBuf, ByteBuf)，它被用来判断两个ByteBuf实例的相等性。

引用计数

Netty为ByteBuf和ByteBufHolder引入了引用计数计数，它们都实现了interface ReferenceCounted。
引用计数对于池化实现来说至关重要，它降低了内存分配的开销。

Channel channel = ...;
ByteufAllocator allocator = channel.alloc();
....
ByteBuf buffer = allocator.directBuffer();
assert buffer.refCnt() == 1;
....
ByteBuf buffer = ...;
boolean released = buffer.release(); //减少到该对象的活动引用。当减少到0时，该对象被释放，并且该方法返回true

ChannelHandler和ChannelPipeline

ChannelHandlerInboundHandler接口

释放消息资源

当某各ChannelInboundHandler的实现重写channelRead()方法时，它将负责显示地释放与池化的ByteBuf实例相关的内存。

Netty将使用WARN级别的日志消息记录未释放的资源，使得可以非常简单地在代码中发现违规的实例。

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;

@ChannelHandler.Sharable
public class DiscardHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //super.channelRead(ctx, msg);
        ReferenceCountUtil.release(msg);
    }
}

由于SimpleChannelInboundHandler会自动释放资源，所以不应该存储指向任何消息的引用供将来使用，因为这些引用都将会失效。

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

@ChannelHandler.Sharable
public class SimpleDiscardHandler extends SimpleChannelInboundHandler {
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
        // No need to do anything special
    }
}

资源管理

每当通过调用ChannelInboundHandler.channelRead()或者ChannelOutboundHandler.write()方法来处理数据时，都需要确保没有任何的资源泄漏。Netty使用引用计数来处理池化的ByteBuf。所以在完全使用完某个ByteBuf后，调整其引用计数是很重要的。

为了帮助你诊断潜在的（资源泄漏）问题，Netty提供了class ResourceLeakDetector，它将对应用程序的缓冲区分配做大约1%的采样来检测内存泄漏。

Netty定义了4种泄漏检测级别

级别	描述
DISABLED	禁用泄漏检测。只有在详尽的测试之后才应设置为这个值
SIMPLE	使用1%的默认采样率检测并报告任何发现的泄漏。这是默认级别，适合绝大部分的情况
ADVANCED	使用默认的采样率，报告所发现的任何的泄漏以及对应的消息被访问的位置
PARANOID	类似于ADVANCED，但是其将会对每次（对消息的）访问都进行采样。这对性能将会有很大的影响，应该只在调试阶段使用

防止泄漏

入站

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;

@ChannelHandler.Sharable
public class DiscardHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //super.channelRead(ctx, msg);
        ReferenceCountUtil.release(msg);
    }
}

出站

不仅要释放资源，还要通知ChannelPromise。否则可能会出现ChannelFutureListener收不到消息已经被处理了的通知的情况。

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.util.ReferenceCountUtil;

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

ChannelHandlerContext接口

ChannelHandlerContext代表了ChannelHandler和ChannelPipeline之间的关联，每当有ChannelHandler添加到ChannelPipeline中时，都会创建ChannelHandlerContext。ChannelHandlerContext的主要功能时管理它所关联的ChannelHandler都在同一个ChannelPipeline中的其他ChannelHandler之间的交互。

使用ChannelHandlerContext

1
2
3

ChannelHandlerContext ctx = ...;
Channel channel = ctx.channel();
channel.write(Unpooled.copiedBuffer("Netty in Action", CharsetUtil.UTF_8));

1
2
3

ChannelHandlerContext ctx = ...;
ChannelPipeline pipeline = ctx.pipeline();
pipeline.write(Unpooled.copiedBuffer("Netty in Action", CharsetUtil.UTF_8));

1 2	ChannelHandlerContext ctx = ...; ctx.write(Unpooled.copiedBuffer("Netty in Action", CharsetUtil.UTF_8));

缓存ChannelHandlerContext的引用

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

public class WriteHandler extends ChannelHandlerAdapter {
    private ChannelHandlerContext ctx;

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        this.ctx = ctx;
    }
    public void send(String msg){
        ctx.writeAndFlush(msg);
    }
}

共享ChannelHandler

因为一个ChannelHandler可以从属于多个ChannelPipeline，所以它也可以绑定到多个ChannelHandlerContext实例；
对于这种用法旨在多个ChannelPipeline中共享一个ChannelHandler，对应的需要注解@Sharable；
否则会触发异常。
为了安全地被用于多个并发的Channel，所以这样的ChannelHandler必须是线程安全的。

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

@ChannelHandler.Sharable
public class SharableHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("Channel read message: " + msg);
        ctx.fireChannelRead(msg);
    }
}

错误用法

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

@ChannelHandler.Sharable
public class SharableHandler extends ChannelInboundHandlerAdapter {
    private int count;
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        count ++;
        System.out.println("ChannelRead(...) called the" + count + " time");
        ctx.fireChannelRead(msg);
    }
}

异常处理

处理入站异常

ChannelHandler.exceptionCaught()的默认实现是简单地将当前异常转发给ChannelPipeline中的下一个ChannelHandler；
如果异常到达了ChannelPipeline的尾端，它将会被记录为未被处理；
要想定义自定义的处理逻辑，需要重写exceptionCaught()方法。然后需要决定是否需要将该异常传播出去。

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class InboundExceptionHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

处理出站异常

用于处理出站操作中的正常完成以及异常的选项，都基于以下的通知机制：

每个出站操作都返回一个ChannelFuture。注册到ChannelFuture的ChannelFutureListener将在操作完成时被通知该操作是成功了还是出错了；
几乎所有的ChannelOutboundHandler上的方法都会传入一个ChannelPromise的实例。作为ChannelFuture的子类，ChannelPromise也可以被分配用于异步通知的监听器。但是，ChannelPromise还具有提供立即通知的可写方法。

方法一

Channel channel = ...
ChannelFuture future = channel.write(someMessage);
future.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        if(!channelFuture.isSuccess()){
            channelFuture.cause().printStackTrace();
            channelFuture.channel().close();
        }
    }
});

方法二

package com.louris.springboot.utils;

import io.netty.channel.*;

public class OutboundExceptionHandler extends ChannelOutboundHandlerAdapter {
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        promise.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                if(channelFuture.isSuccess()){
                    channelFuture.cause().printStackTrace();
                    channelFuture.channel().close();
                }
            }
        });
    }
}

EventLoop和线程模型

任务调度

JDK的任务调度API

ScheduledExecutorService executor = Executors.newScheduledThreadPool(10);

ScheduledFuture<?> future = executor.schedule(new Runnable() {
    @Override
    public void run() {
        System.out.println("60 seconds later");
    }
}, 60, TimeUnit.SECONDS);

executor.shutdown();

使用EventLoop调度任务

Channel ch = ...
//ScheduledFuture<?> future = ch.eventLoop().schedule(new Runnable() {
//    @Override
//    public void run() {
//        System.out.println("60 seconds later");
//    }
//}, 60, TimeUnit.SECONDS);
ScheduledFuture<?> future = ch.eventLoop().scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        System.out.println("60 seconds later");
    }
}, 60, 60, TimeUnit.SECONDS);
boolean mayInterruptIfRunning = false;
future.cancel(mayInterruptIfRunning);

引导

引导客户端

EventLoopGroup group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
        .channel(NioSocketChannel.class)
        .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("www.manning.com", 80));
future.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        if(channelFuture.isSuccess()){
            System.out.println("Connection established");
        }else{
            System.out.println("Connection attempt failed");
            channelFuture.cause().printStackTrace();
        }
    }
});

Channel和EventLoopGroup的兼容性

Nio前缀与Oio前缀的不能混淆使用

引导服务器

常用方式

NioEventLoopGroup group = new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
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");
            }
        });
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("Bound attempt failed");
            channelFuture.cause().printStackTrace();
        }
    }
});

从Channel引导到客户端

假设你的服务器正在处理一个客户端的请求，这个请求需要它充当第三方系统的客户端；
当一个应用程序（如一个代理服务器）必须要和组织现有的系统（如Web服务或者数据库）集成时，就可能发生这种情况；
在这种情况下，将需要从已经被接受的子Channel中引导一个客户端Channel；
可以按照前一节创建新的Bootstrap实例，但是这并不是最高效的解决方案，因为它将要求你为每个新创建的客户端Channel定义另一个EventLoop。这会产生额外的线程，以及在已被接受的子Channel和客户端Channel之间交换数据时不可避免的上下文切换；
更好的解决方案是：通过将已被接受的子Channel的EventLoop传递给Bootstrap的group()方法来共享该EventLoop。因为分配给EventLoop的所有Channel都使用同一个线程，所以这避免了额外的线程创建。

ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(new NioEventLoopGroup(), new NioEventLoopGroup())
        .channel(NioServerSocketChannel.class)
        .childHandler(new SimpleChannelInboundHandler<ByteBuf>() {
            ChannelFuture connectFuture;

            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.channel(NioServerSocketChannel.class)
                        .handler(new SimpleChannelInboundHandler<ByteBuf>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
                                System.out.println("Received data");
                            }
                        });
                bootstrap.group(ctx.channel().eventLoop());
                connectFuture = bootstrap.connect(new InetSocketAddress("www.manning.com", 8080));
            }

            @Override
            protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
                if(connectFuture.isDone()){
                    // do something with the data
                }
            }
        });
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();
        }
    }
});

在引导过程中添加多个ChannelHandler

ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(new NioEventLoopGroup(), new NioEventLoopGroup())
        .channel(NioServerSocketChannel.class)
        .childHandler(new ChannelInitializerImpl());
ChannelFuture future = bootstrap.bind(new InetSocketAddress(8080));
future.sync();

final static class ChannelInitializerImpl extends ChannelInitializer<Channel>{
    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        pipeline.addLast(new HttpClientCodec());
        pipeline.addLast(new HttpObjectAggregator(Integer.MAX_VALUE));
    }
}

使用Netty的ChannelOption和属性

final AttributeKey<Integer> id = AttributeKey.newInstance("ID");  // 创建一个AttributeKey以标识该属性
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(new NioEventLoopGroup())
        .channel(NioServerSocketChannel.class)
        .handler(new SimpleChannelInboundHandler<ByteBuf>() {
            @Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                Integer idValue = ctx.channel().attr(id).get();
                // do something with the idValue
            }

            @Override
            protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
                System.out.println("Received data");
            }
        });
bootstrap.option(ChannelOption.SO_KEEPALIVE, true)
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);  // 设置ChannelOption，其将在connect()或者bind()方法被调用时被设置到已经创建的Channel上
bootstrap.attr(id, 123456); // 存储该id属性
ChannelFuture future = bootstrap.connect(new InetSocketAddress("www.manning.com", 80));
future.syncUninterruptibly();

引导DatagramChannel

前面的引导代码都是基于TCP协议的SocketChannel，但是Bootstrap类也可以被用于无连接的协议。
为此，Netty提供了各种DatagramChannel的实现;
唯一的区别是，不再调用connect()方法，而是只调用bind()方法

Bootstrap bootstrap = new Bootstrap();
bootstrap.group(new OioEventLoopGroup())
        .channel(OioDatagramChannel.class)
        .handler(new SimpleChannelInboundHandler<DatagramPacket>() {
            @Override
            protected void channelRead0(ChannelHandlerContext channelHandlerContext, DatagramPacket datagramPacket) throws Exception {
                // Do something with the packet
            }
        });
ChannelFuture future = bootstrap.bind(new InetSocketAddress(0));
future.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        if(channelFuture.isSuccess()){
            System.out.println("Channel bound");
        }else{
            System.out.println("Bind attempt failed");
            channelFuture.cause().printStackTrace();
        }
    }
});

关闭

最重要的是，需要关闭EventLoopGroup，它将处理任何挂起的事件和任务，并且随后释放所有活动的线程：EventLoopGroup.shutdownGracefully()方法的作用；
shutdownGracefully()也是一个异步的操作，所以需要阻塞等待直到它完成，或者向所返回的Future注册一个监听器以在关闭完成时获得通知；
也可以在调用shutdowngracefully()方法之前，显示地在所有活动的Channel上调用Channel.close()方法。
当时在任何情况下，都得记得关闭EventLoopGroup本身

EventLoopGroup group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
        .channel(NioSocketChannel.class);
...
Future<?> future = group.shutdownGracefully();  // 释放所有资源，并且关闭所有的当前正在使用的Channel
// block until the group has shutdown
future.syncUniterruptibly();

单元测试

EmbeddedChannel概述

名称	职责
writeInbound(Object… msgs)	将入站消息写到EmbeddedChannel中。如果可以通过readInbound()方法从EmbeddedChannel中读取数据，则返回true
readInbound()	从EmbeddedChannel中读取一个入站消息。任何返回的东西都穿越了整个ChannelPipeline。如果没有任何可供读取的，则返回null
writeOutbound(Object… msgs)	将出站消息写到EmbeddedChannel中。如果现在可以通过readOutbound()方法从EmbeddedChannel中读取到什么东西，则返回true
readOutbound()	从EmbeddedChannel中读取一个出站消息。任何返回的东西都穿越了整个ChannelPipeline。如果没有任何可供读取的，则返回null
finish()	将EmbeddedChannel标记为完成，并且如果可被读取的入站数据或者出站数据，则返回true。这个方法还将会调用EmbeddedChannel上的close()方法

使用EmbeddedChannel测试ChannelHandler

测试入站消息

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

public class FixedLengthFrameDecoder extends ByteToMessageDecoder {
    private final int frameLength;

    public FixedLengthFrameDecoder(int frameLength){
        if(frameLength <= 0){
            throw new IllegalArgumentException("frameLength must be a positive integer: " + frameLength);
        }
        this.frameLength = frameLength;
    }

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
        while(in.readableBytes() >= frameLength){
            ByteBuf buf = in.readBytes(frameLength);
            out.add(buf);
        }
    }
}

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.embedded.EmbeddedChannel;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.springframework.util.Assert;


public class FixedLengthFrameDecoderTest {
    @Test
    public void testFramesDecoded(){
        ByteBuf buf = Unpooled.buffer();
        for(int i = 0; i < 9; i ++){
            buf.writeByte(i);
        }
        ByteBuf input = buf.duplicate();
        EmbeddedChannel channel = new EmbeddedChannel(new FixedLengthFrameDecoder(3));
        // write bytes
        Assertions.assertTrue(channel.writeInbound(input.retain()));
        Assertions.assertTrue(channel.finish());

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

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

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

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

    @Test
    public void testFramesDecoded2(){
        ByteBuf buf = Unpooled.buffer();
        for(int i = 0; i < 9; i ++){
            buf.writeByte(i);
        }
        ByteBuf input = buf.duplicate();
        EmbeddedChannel channel = new EmbeddedChannel(new FixedLengthFrameDecoder(3));
        // write bytes
        Assertions.assertTrue(channel.writeInbound(input.readBytes(2)));  // 返回false，因为没有一个完整的可供读取的帧
        Assertions.assertTrue(channel.writeInbound(input.readBytes(7)));
        Assertions.assertTrue(channel.finish());

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

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

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

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

测试出站消息

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageEncoder;

import java.util.List;

public class AbsIntegerEncoder extends MessageToMessageEncoder<ByteBuf> {
    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
        while(in.readableBytes() >= 4){
            int value = Math.abs(in.readInt());
            out.add(value);
        }
    }
}

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.embedded.EmbeddedChannel;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

public class AbsIntegerEncoderTest {
    @Test
    public void testEncoded(){
        ByteBuf buf = Unpooled.buffer();
        for(int i = 1; i < 10; i ++){
            buf.writeInt(i * -1);
        }
        EmbeddedChannel channel = new EmbeddedChannel(new AbsIntegerEncoder());

        Assertions.assertTrue(channel.writeOutbound(buf));
        Assertions.assertTrue(channel.finish());

        // read bytes
        for(int i = 1; i < 10; i ++){
            Assertions.assertEquals(i, (int)channel.readOutbound());
        }
        Assertions.assertNull(channel.readOutbound());
    }
}

测试异常处理

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.TooLongFrameException;

import java.util.List;

public class FrameChunkDecoder extends ByteToMessageDecoder {
    private final int maxFrameSize;
    
    public FrameChunkDecoder(int maxFrameSize){
        this.maxFrameSize = maxFrameSize;
    }

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
        int readableBytes = in.readableBytes();
        if(readableBytes > maxFrameSize){  // 如果该帧太大，则丢弃它并抛出一个异常
            // discard the bytes
            in.clear();
            throw new TooLongFrameException();
        }
        ByteBuf buf = in.readBytes(readableBytes);
        out.add(buf);
    }
}

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.TooLongFrameException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;

public class FrameChunkDecoderTest {
    @Test
    public void testFramesDecoded(){
        ByteBuf buf = Unpooled.buffer();
        for(int i = 0; i < 9; i ++){
            buf.writeByte(i);
        }

        ByteBuf input = buf.duplicate();

        EmbeddedChannel channel = new EmbeddedChannel(new FrameChunkDecoder(3));

        Assertions.assertTrue(channel.writeInbound(input.readBytes(2)));
        try{
            channel.writeInbound(input.readBytes(4));
            Assertions.fail();
        }catch (TooLongFrameException e){
            // expected exception
            System.out.println("Exception happen!");
        }

        Assertions.assertTrue(channel.writeInbound(input.readBytes(3)));
        Assertions.assertTrue(channel.finish());

        // Read frames
        ByteBuf read = (ByteBuf) channel.readInbound();
        Assertions.assertEquals(buf.readSlice(2), read);
        read.release();

        read = (ByteBuf) channel.readInbound();
        Assertions.assertEquals(buf.skipBytes(4).readSlice(3), read);
        read.release();
        buf.release();
    }
}

编解码器框架

编解码器：每个网络应用程序都必须定义如何解析在两个节点之间来回传输的原始字节，以及如何将其和目标应用程序的数据格式做相互转换。这种转换逻辑由编解码器处理，编解码器由编码器和解码器组成，它们每种都可以将字节流从一种格式转换为另一种格式；
编码器是将消息转换为适合用于传输的格式（最有可能的就是字节流）；
解码器是将网络字节流转换回应用程序的消息个事；
因此，编码器操作出栈数据，而解码器处理入站数据。

解码器

ByteToMessageDecoder

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

public class ToIntegerDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
        if(in.readableBytes() >= 4){
            out.add(in.readInt());
        }
    }
}

编解码器中的引用计数：

一旦消息被解码或者编码，它就会被ReferenceCountUtil.release(message) 调用自动释放；
如果需要保留引用以便稍后使用，那么可以调用ReferenceCountUtil.retain(message)方法，这将会增加该引用计数，从而防止该消息被释放。

ReplayingDecoder

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;

import java.util.List;

public class ToIntegerDecoder2 extends ReplayingDecoder<Void> {
    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
        out.add(in.readInt());
    }
}

如果没有足够的字节可用，这个readInt()方法的实现将抛出一个Error，其将在基类中被捕获并处理；
当有更多的数据可供读取时，该decode()方法将会被再次调用；
并不是所有的ByteBuf操作都被支持，如果调用了一个不被支持的方法，将会抛出一个UnsupportedOperationException；
ReplayingDecoder稍慢于ByteToMessageDecoder。

更多的解码器

LineBasedFrameDecoder;
HttpObjectDecoder;
io.netty.handler.codec子包下面，将会发现更多用于特定用例的编码器和解码器实现。

MessageToMessageDecoder

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;

import java.util.List;

public class IntegerToStringDecoder extends MessageToMessageDecoder<Integer> {
    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, Integer msg, List<Object> out) throws Exception {
        out.add(String.valueOf(msg));
    }
}

TooLongFrameException

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.TooLongFrameException;

import java.util.List;

public class SafeByteToMessageDecoder extends ByteToMessageDecoder {
    private static final int MAX_FRAME_SIZE = 1024;

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, 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!");
        }
        
        // do something
        
    }
}

编码器

将消息编码为字节
将消息编码为消息

MessageToByteEncoder

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

public class ShortToByteEncoder extends MessageToByteEncoder<Short> {
    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, Short msg, ByteBuf out) throws Exception {
        out.writeShort(msg);
    }
}

MessageToMessageEncoder

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageEncoder;

import java.util.List;

public class IntegerToStringEncoder extends MessageToMessageEncoder<Integer> {
    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, Integer msg, List<Object> out) throws Exception {
        out.add(String.valueOf(msg));
    }
}

抽象编解码器类

在同一个类中管理入站和出战数据和消息的转换

抽象类ByteToMessageCodec

它结合了ByteToMessageDecoder以及它的逆向——MessageToByteEncoder；
任何的请求/响应协议都可以作为使用ByteToMessageCodec的理想选择

抽象类MessageToMessageCodec

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import io.netty.handler.codec.http.websocketx.*;
import org.w3c.dom.Text;

import java.util.List;

public class WebSocketConvertHandler extends MessageToMessageCodec<WebSocketFrame, WebSocketConvertHandler.MyWebSocketFrame>{

    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, MyWebSocketFrame msg, List<Object> out) throws Exception {
        ByteBuf payload = msg.getData().duplicate().retain();
        switch (msg.getType()){
            case BINARY:
                out.add(new BinaryWebSocketFrame(payload));
                break;
            case TEXT:
                out.add(new TextWebSocketFrame(payload));
                break;
            case CLOSE:
                out.add(new CloseWebSocketFrame(true, 0, payload));
                break;
            case CONTINUATION:
                out.add(new ContinuationWebSocketFrame(payload));
                break;
            case PONG:
                out.add(new PongWebSocketFrame(payload));
                break;
            case PING:
                out.add(new PingWebSocketFrame(payload));
                break;
            default:
                throw new IllegalArgumentException("Unsupported websocket msg " + msg);
        }
    }

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, WebSocketFrame msg, List<Object> out) throws Exception {
        ByteBuf payload = msg.content().duplicate().retain();
        if(msg instanceof BinaryWebSocketFrame){
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.BINARY, payload));
        }else if(msg instanceof CloseWebSocketFrame){
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.CLOSE, payload));
        }else if(msg instanceof PingWebSocketFrame){
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.PING, payload));
        }else if(msg instanceof PongWebSocketFrame){
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.PONG, payload));
        }else if(msg instanceof TextWebSocketFrame){
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.TEXT, payload));
        }else if(msg instanceof ContinuationWebSocketFrame){
            out.add(new MyWebSocketFrame(MyWebSocketFrame.FrameType.CONTINUATION, payload));
        }else{
            throw new IllegalArgumentException("Unsupported websocket msg " + msg);
        }
    }

    public static final class MyWebSocketFrame {
        public enum FrameType {
            BINARY,
            CLOSE,
            PING,
            PONG,
            TEXT,
            CONTINUATION
        }

        private final FrameType type;
        private final ByteBuf data;

        public MyWebSocketFrame(FrameType type, ByteBuf data) {
            this.type = type;
            this.data = data;
        }

        public FrameType getType() {
            return type;
        }

        public ByteBuf getData() {
            return data;
        }
    }
}

CombinedChannelDuplexHandler类

结合一个解码器和编码器可能会对可重用性造成影响；
CombinedChannelDuplexHandler类即能避免这种惩罚，又不会牺牲将一个解码器和一个编码器作为一个单独的单元部署所带来的的便利性

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

public class ByteToCharDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
        while(in.readableBytes() >= 2){
            out.add(in.readChar());
        }
    }
}

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

public class CharToByteEncoder extends MessageToByteEncoder<Character> {
    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, Character msg, ByteBuf out) throws Exception {
        out.writeChar(msg);
    }
}

package com.louris.springboot.utils;

import io.netty.channel.CombinedChannelDuplexHandler;

public class CombinedByteCharCodec extends CombinedChannelDuplexHandler<ByteToCharDecoder, CharToByteEncoder> {
    public CombinedByteCharCodec(){
        super(new ByteToCharDecoder(), new CharToByteEncoder());
    }
}

预置的ChannelHandler和编解码器

通过SSL/TLS保护Netty应用程序

为了支持SSL/TLS，Java提供了javax.net.ssl包，它的SSLContext和SSLEngine类使得实现解密和加密相当简单直接；
Netty通过一个名为SslHandler的ChannelHandler实现利用了这个API，其中SslHandler在内部使用SSLEngine来完成实际的工作。

package com.louris.springboot.utils;

import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;

import javax.net.ssl.SSLEngine;

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 channel) throws Exception {
        SSLEngine engine = context.newEngine(channel.alloc());
        channel.pipeline().addFirst("ssl", new SslHandler(engine, startTls));
    }
}

构建基于Netty的HTTP/HTTPS应用程序

HTTP解编码器

package com.louris.springboot.utils;

import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.codec.http.HttpRequestEncoder;
import io.netty.handler.codec.http.HttpResponseDecoder;
import io.netty.handler.codec.http.HttpResponseEncoder;

public class HttpPipelineInitializer extends ChannelInitializer<Channel> {
    private final boolean client;

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

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

聚合HTTP消息

由于HTTP的请求和响应可能由许多部分组成，它可以将多个消息部分合并为FullHttpRequest或者FullHttpResponse消息。
由于消息分段需要被缓冲，直到可以转发一个完整的消息给下一个ChannelInboundHandler，所以这个操作有轻微的开销。其带来的好处便是不不必关心消息碎片了。

package com.louris.springboot.utils;

import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;

public class HttpAggregatorInitializer extends ChannelInitializer<Channel> {
    private final boolean isClient;

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

    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        if(isClient){
            pipeline.addLast("codec", new HttpClientCodec());
        }else{
            pipeline.addLast("codec", new HttpServerCodec());
        }
        // 将最大消息大小为512KB的HttpObjectAggregator添加到ChannelPipeline
        pipeline.addLast("aggregator", new HttpObjectAggregator(512 * 1024));
    }
}

HTTP压缩

当使用HTTP时，建议开启压缩功能以尽可能多地减少传输数据的大小；
虽然压缩会带来一些CPU时钟周期上的开销，但是通常来说它都是一个好主意，特别是对于文本数据来说；
Netty同时支持gzip和deflate编码的压缩实现。

1
2
3

GET /encrypted-area HTTP/1.1
Host: www.example.com
Accept-Encoding: gzip, deflate

package com.louris.springboot.utils;

import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpContentCompressor;
import io.netty.handler.codec.http.HttpContentDecompressor;
import io.netty.handler.codec.http.HttpServerCodec;

public class HttpCompressionInitializer extends ChannelInitializer<Channel> {
    private final boolean isClient;
    
    public HttpCompressionInitializer(boolean isClient){
        this.isClient = isClient;
    }

    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        if(isClient){
            pipeline.addLast("codec", new HttpClientCodec());
            pipeline.addLast("decompressor", new HttpContentDecompressor());
        }else{
            pipeline.addLast("codec", new HttpServerCodec());
            pipeline.addLast("compressor", new HttpContentCompressor());
        }
    }
}

使用HTTPS

package com.louris.springboot.utils;

import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;

import javax.net.ssl.SSLEngine;

public class HttpsCodecInitializer extends ChannelInitializer<Channel> {
    private final SslContext context;
    private final boolean isClient;

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

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

        if(isClient){
            pipeline.addLast("codec", new HttpClientCodec());
        }else{
            pipeline.addLast("codec", new HttpServerCodec());
        }
    }
}

WebSocket

WebSocket在客户端和服务器之间提供了真正的双向数据交换；
WebSocketFrame分为数据帧和控制帧；

名称	描述
BinaryWebSocketFrame	数据帧：二进制数据
TextWebSocketFrame	数据帧：文本数据
ContinuationWebSocketFrame	数据帧：属于上一个BinaryWebSocketFrame或者TextWebSocketFrame的文本的或者二进制数据
CloseWebSocketFrame	控制帧：一个CLOSE请求、关闭的状态码以及关闭的原因
PingWebSocketFrame	控制帧：请求一个PongWebSocketFrame
PongWebSocketFrame	控制帧：对PingWebSocketFrame请求的响应

package com.louris.springboot.utils;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;

public class WebSocketServerInitializer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel channel) throws Exception {
        channel.pipeline().addLast(
                new HttpServerCodec(),
                new HttpObjectAggregator(65536),
                new WebSocketServerProtocolHandler("/websocket"),
                new TextFrameHandler(),
                new BinaryFrameHandler(),
                new ContinuationFrameHandler());
    }

    public static final class TextFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame>{
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {
            // handle text frame
        }
    }
    public static final class BinaryFrameHandler extends SimpleChannelInboundHandler<BinaryWebSocketFrame>{
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, BinaryWebSocketFrame binaryWebSocketFrame) throws Exception {
            // handle binary frame
        }
    }

    public static final class ContinuationFrameHandler extends SimpleChannelInboundHandler<ContinuationFrameHandler>{
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, ContinuationFrameHandler continuationFrameHandler) throws Exception {
            // handle continuation frame
        }
    }
}

解码基于分隔符的协议和基于长度的协议

在使用Netty的过程中，将会遇到需要解码器的基于分隔符和帧长度的协议。

基于分隔符的协议

名称	描述
DelimiterBasedFrameDecoder	使用任何由用户提供的分隔符来提取帧的通用解码器
LineBasedFrameDecoder	提取由行尾符（\n或者\r\n）分隔的帧的解码器。这个解码器比上面这个更快

行尾分隔符示例

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.handler.codec.LineBasedFrameDecoder;

public class LineBasedHandlerInitializer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        pipeline.addLast(new LineBasedFrameDecoder(64 * 1024));  //将提取的帧转发给下一个ChannelInboundHandler
        pipeline.addLast(new FrameHandler()); //添加FrameHandler以接收帧
    }
    
    public static final class FrameHandler extends SimpleChannelInboundHandler<ByteBuf>{
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
            // do something with the data extracted from the frame
        }
    }
}

如果使用除了行尾符之外的分隔符分隔的帧，那么可以以类似的方式使用DelimiterBasedFrameDecoder，只需要将特定的分隔符序列指定到其构造函数即可。

自定义空格分隔符

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.handler.codec.LineBasedFrameDecoder;

public class CmdHandlerInitializer extends ChannelInitializer<Channel> {
    static final byte SPACE = (byte) ' ';

    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        pipeline.addLast(new CmdDecoder(64 * 1024));
        pipeline.addLast(new CmdHandler());
    }

    public static final class Cmd{  //将帧的内容存储到ByteBuf上
        private final ByteBuf name;
        private final ByteBuf args;

        public Cmd(ByteBuf name, ByteBuf args){
            this.name = name;
            this.args = args;
        }

        public ByteBuf getName() {
            return name;
        }

        public ByteBuf getArgs() {
            return args;
        }
    }

    public static final class CmdDecoder extends LineBasedFrameDecoder{
        public CmdDecoder(int maxLength){
            super(maxLength);
        }

        @Override
        protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception {
            ByteBuf frame = (ByteBuf) super.decode(ctx, buffer);
            if(frame == null) return null;
            int index = frame.indexOf(frame.readerIndex(), frame.writerIndex(), SPACE); // 查找第一个空格字符的索引。前面是命令名称，接着是参数
            return new Cmd(frame.slice(frame.readerIndex(), index),
                    frame.slice(index + 1, frame.writerIndex()));
        }
    }

    public static final class CmdHandler extends SimpleChannelInboundHandler<Cmd>{
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, Cmd cmd) throws Exception {
            // do something with the command
        }
    }
}

基于长度的协议

名称	描述
FixedLengthFrameDecoder	提取在调用构造函数时指定的定长帧
LengthFieldBasedFrameDecoder	根据编码进帧头部中的长度值提取帧；该字段的偏移量以及长度在构造函数中指定

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;

public class LengthBasedInitializer extends ChannelInitializer<Channel> {
    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        // 帧的长度编码到帧起始的前8个字节中的消息
        pipeline.addLast(new LengthFieldBasedFrameDecoder(64 * 1024, 0, 8));
        pipeline.addLast(new FrameHandler());
    }
    
    public static final class FrameHandler extends SimpleChannelInboundHandler<ByteBuf>{
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
            // do something with the frame
        }
    }
}

写大型数据

因为网络饱和的可能性，如何在异步框架中高效地写大块的数据是一个特殊的问题；
由于写操作时非阻塞的，所以即使没有写出所有的数据，写操作也会在完成时返回并通知ChannelFuture；
当这种情况发生时，如果仍然不停地写入，就有内存耗尽的风险；
所以在写大型数据时，需要准备好处理到远程节点的连接是慢速连接的情况，这种情况会导致内存释放的延迟。

零拷贝特性：

所有这一切都发生在Netty的核心中，所以应用程序所有需要做的就是使用一个FileRegion接口的实现，其在Netty的API文档中的定义是：“通过支持零拷贝的文件传输的Channel来发送的文件区域”

以下代码展示了如何通过从FileInputStream创建一个DefaultFileRegion，并将其写入Channel，从而利用零拷贝特性来传输一个文件的内容：

File file = ...
Channel channel = ...
FileInputStream in = new FileInputStream(file);
FileRegion region = new DefaultFileRegion(in.getChannel(), 0, file.length());
channel.writeAndFlush(region).addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        if(!channelFuture.isSuccess()){
            Throwable cause = channelFuture.cause();
            // do something
        }
    }
});

当Channel的状态变为活动时，WriteStreamHandler将会逐块地把来自文件中的数据作为ChunkedStream写入。数据在传输之前将会由SslHandler加密。

package com.louris.springboot.utils;

import io.netty.channel.*;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedStream;
import io.netty.handler.stream.ChunkedWriteHandler;

import java.io.File;
import java.io.FileInputStream;

public class ChunkedWriteHandlerInitializer extends ChannelInitializer<Channel> {
    private final File file;
    private final SslContext sslCtx;

    public  ChunkedWriteHandlerInitializer(File file, SslContext sslCtx){
        this.file = file;
        this.sslCtx = sslCtx;
    }

    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        pipeline.addLast(new SslHandler(sslCtx.newEngine(channel.alloc())));
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new WriteStreamHandler());
    }
    
    public final class WriteStreamHandler extends ChannelInboundHandlerAdapter{
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            super.channelActive(ctx);
            ctx.writeAndFlush(new ChunkedStream(new FileInputStream(file)));
        }
    }
}

序列化数据

JDK提供了ObjectOutputStream和ObjectInputStream，用于通过网络对POJO的的基本数据类型和图进行序列化和反序列化；
该API并不复杂，而且可以被应用于任何实现了java.io.Serializable接口的对象；
但是它的性能也不是非常搞笑的。

JDK序列化

JDK序列化编码器

名称	描述
CompatibleObjectDecoder	和使用JDK序列化的非基于Netty的远程节点进行互操作的解码器
CompatibleObjectEncoder	和使用JDK序列化的非基于Netty的远程节点进行互操作的编码器
ObjectDecoder	构建于JDK序列化之上的使用自定义的序列化来解码的解码器；当没有其他的外部依赖时，它提供了速度上的改进。否则其他的序列化实现更加可取
ObjectEncoder	构建于JDK序列化之上的使用自定义的序列化来编码的编码器；当没有其他的外部依赖时，它提供了速度上的改进。否则其他的序列化实现更加可取

使用JBoss Marshalling进行序列化

如果可以自由地使用外部依赖，那么JBoss Marshalling将是个理想的选择：它比JDK序列化最多快3倍，而且也更加紧凑；
Netty通过下表所示的两组解码器/编码器对为Boss Marshalling提供了支持。
第一组兼容只使用JDK序列化的远程节点；
第二组提供了最大的性能，适用于和使用JBoss Marshalling的远程节点一起使用

名称	描述
CompatibleMarshallingDecoder	与只使用JDK序列化的远程节点兼容
CompatibleMarshallingEncoder
MarshallingDecoder	适用于使用JBoss Marshalling的节点。这些类必须一起使用
MarshallingEncoder

package com.louris.springboot.utils;

import io.netty.channel.*;
import io.netty.handler.codec.marshalling.MarshallerProvider;
import io.netty.handler.codec.marshalling.MarshallingDecoder;
import io.netty.handler.codec.marshalling.MarshallingEncoder;
import io.netty.handler.codec.marshalling.UnmarshallerProvider;

import java.io.Serializable;

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();
        pipeline.addLast(new MarshallingDecoder(unmarshallerProvider)); //将ByteBuf转换为POJO
        pipeline.addLast(new MarshallingEncoder(marshallerProvider));  // 将POJO转换为ByteBuf
        pipeline.addLast(new ObjectHandler()); // ObjectHandler处理普通的实现了Serializable接口的POJO
    }

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

通过Protocol Buffers序列化

名称	描述
ProtobufDecoder	使用protobuf对消息进行解码
ProtobufEncoder	使用protobuf对消息进行编码
ProtobufVarint32FrameDecoder	根据消息中的Google Protocol Buffers的”Base 128 Varints”整型长度字段值动态地分割所接收的ByteBuf
ProtobufVarint32LengthFieldPrepender	向ByteBuf前追加一个Google Protocal Buffers的”Base 128 Varints”整型的长度字段值

添加依赖

<dependency>
    <groupId>com.google.protobuf</groupId>
    <artifactId>protobuf-java</artifactId>
    <version>3.6.1</version>
</dependency>
<dependency>
    <groupId>com.googlecode.protobuf-java-format</groupId>
    <artifactId>protobuf-java-format</artifactId>
    <version>1.4</version>
</dependency>

package com.louris.springboot.utils;

import com.google.protobuf.MessageLite;
import io.netty.channel.*;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;

public class ProtoBufInitializer extends ChannelInitializer<Channel> {
    private final MessageLite lite;

    public ProtoBufInitializer(MessageLite lite){
        this.lite = lite;
    }

    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        pipeline.addLast(new ProtobufVarint32FrameDecoder());
        pipeline.addLast(new ProtobufEncoder());
        pipeline.addLast(new ProtobufDecoder(lite));
        pipeline.addLast(new ObjectHandler());
    }
    
    public static final class ObjectHandler extends SimpleChannelInboundHandler<Object>{
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
            // do something
        }
    }
}

WebSocket

添加WebSocket支持

处理HTTP请求

如果地址为/ws的URI被访问，那么服务器将会处理WebSocket升级

package com.louris.springboot.utils;

import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedNioFile;

import java.io.File;
import java.io.RandomAccessFile;
import java.net.URISyntaxException;
import java.net.URL;
import java.util.Random;

public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
    private final String wsUri;
    private static final File INDEX;

    static {
        URL location = HttpRequestHandler.class
                .getProtectionDomain()
                .getCodeSource()
                .getLocation();
        try{
            String path = location.toURI() + "index.html";
            path = !path.contains("file:") ? path : path.substring(5);
            INDEX = new File(path);
        }catch (URISyntaxException e){
            throw new IllegalArgumentException("Unable to locate index.html", e);
        }
    }

    public HttpRequestHandler(String wsUri){
        this.wsUri = wsUri;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
        if(wsUri.equalsIgnoreCase(request.getUri())){ // 如果请求了WebSocket协议升级，则增加引用计数（调用retrain()方法），并将它传递给下一个ChannelInboundHandler
            ctx.fireChannelRead(request.retain());
        }else{
            if(HttpHeaders.is100ContinueExpected(request)){ //处理100个Continue请求以符合HTTP1.1规范
                send100Continue(ctx);
            }
            RandomAccessFile file = new RandomAccessFile(INDEX, "r"); //读取index.html
            HttpResponse response = new DefaultHttpResponse(
                    request.getProtocolVersion(),
                    HttpResponseStatus.OK
            );
            response.headers().set(
                    HttpHeaders.Names.CONTENT_TYPE,
                    "text/plain; charset=UTF-8"
            );
            boolean keepAlive = HttpHeaders.isKeepAlive(request);
            if(keepAlive){
                response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, file.length());
                response.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
            }
            ctx.write(response);  // 将HttpResponse写到客户端
            if(ctx.pipeline().get(SslHandler.class) == null){  // 将index.html写到客户端
                ctx.write(new DefaultFileRegion(file.getChannel(), 0, file.length()));
            }else{
                ctx.write(new ChunkedNioFile(file.getChannel()));
            }
            ChannelFuture future = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT); // 写LastHttpContent并冲刷至客户端
            if(!keepAlive){  // 如果没有请求keep-alive，则在写操作完成后关闭Channel
                future.addListener(ChannelFutureListener.CLOSE);
            }
        }
    }

    private static void send100Continue(ChannelHandlerContext ctx){
        FullHttpResponse response = new DefaultFullHttpResponse(
                HttpVersion.HTTP_1_1,
                HttpResponseStatus.CONTINUE
        );
        ctx.writeAndFlush(response);
    }

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

处理WebSocket帧

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;

public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    private final ChannelGroup group;

    public TextWebSocketFrameHandler(ChannelGroup group){
        this.group = group;
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if(evt == WebSocketServerProtocolHandler
        .ServerHandshakeStateEvent.HANDSHAKE_COMPLETE){
            ctx.pipeline().remove(HttpRequestHandler.class); // 如果该时间表示握手成功，则从该ChannelPipeline中移出该Handler，因为将不会接接收到任何HTTP消息了
            group.writeAndFlush(new TextWebSocketFrame(
                    "Client" + ctx.channel() + " joined"
            )); // 通知所有已经连接的WebSocket客户端新的客户端已经连接上了
            group.add(ctx.channel());  // 将新的WebSocket Channel添加到ChannelGroup中，以便它可以接收到所有的消息
        }else{
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        group.writeAndFlush(msg.retain());  // 增加消息的引用计数，并将它写到ChannelGroup中所有已经连接的客户端
    }
}

初始化ChannelPipeline

package com.louris.springboot.utils;

import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

public class ChatServerInitializer  extends ChannelInitializer<Channel> {

    private final ChannelGroup group;

    public ChatServerInitializer(ChannelGroup group){
        this.group = group;
    }

    @Override
    protected void initChannel(Channel channel) throws Exception {
        ChannelPipeline pipeline = channel.pipeline();
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new HttpObjectAggregator(64 * 1024));
        pipeline.addLast(new HttpRequestHandler("/ws"));
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
        pipeline.addLast(new TextWebSocketFrameHandler(group));
    }
}

引导

package com.louris.springboot.utils;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.ImmediateEventExecutor;

import java.net.InetSocketAddress;

public class ChatServer {

    private final ChannelGroup channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);

    private final EventLoopGroup group = new NioEventLoopGroup();

    private Channel channel;

    public ChannelFuture start(InetSocketAddress address){
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(group)
                .channel(NioServerSocketChannel.class)
                .childHandler(createInitializer(channelGroup));
        ChannelFuture future = bootstrap.bind(address);
        future.syncUninterruptibly();
        channel = future.channel();
        return future;
    }

    protected ChannelInitializer<Channel> createInitializer(ChannelGroup group){
        return new ChatServerInitializer(group);
    }

    public void destropy(){
        if(channel != null){
            channel.close();
        }
        channelGroup.close();
        group.shutdownGracefully();
    }

    public static void main(String[] args) throws Exception{
        if(args.length != 1){
            System.err.println("Please give port as argument");
            System.exit(1);
        }
        int port = Integer.parseInt(args[0]);
        final ChatServer endpoint = new ChatServer();
        ChannelFuture future = endpoint.start(new InetSocketAddress(port));
        Runtime.getRuntime().addShutdownHook(new Thread(){
            public void run(){
                endpoint.destropy();
            }
        });
        future.channel().closeFuture().syncUninterruptibly();
    }
}

加密

package com.louris.springboot.utils;

import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslHandler;

import javax.net.ssl.SSLEngine;

public class SecureChatServerInitializer extends ChatServerInitializer{

    private final SslContext context;

    public SecureChatServerInitializer(ChannelGroup group, SslContext context){
        super(group);
        this.context = context;
    }

    @Override
    protected void initChannel(Channel channel) throws Exception {
        super.initChannel(channel);
        SSLEngine engine = context.newEngine(channel.alloc());
        engine.setUseClientMode(false);
        channel.pipeline().addFirst(new SslHandler(engine));
    }
}

package com.louris.springboot.utils;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.util.SelfSignedCertificate;

import java.net.InetSocketAddress;

public class SecureChatServer extends ChatServer{
    private final SslContext context;

    public SecureChatServer(SslContext context){
        this.context = context;
    }

    @Override
    protected ChannelInitializer<Channel> createInitializer(ChannelGroup group) {
        return new SecureChatServerInitializer(group, context);
    }

    public static void main(String[] args) throws Exception{
        if(args.length != 1){
            System.err.println("Please give port as argument");
            System.exit(1);
        }

        int port = Integer.parseInt(args[0]);
        SelfSignedCertificate cert = new SelfSignedCertificate();
        SslContext context = SslContext.newServerContext(cert.certificate(), cert.privateKey());

        final SecureChatServer endpoint = new SecureChatServer(context);
        ChannelFuture future = endpoint.start(new InetSocketAddress(port));
        Runtime.getRuntime().addShutdownHook(new Thread(){
            @Override
            public void run() {
                endpoint.destropy();
            }
        });
        future.channel().closeFuture().syncUninterruptibly();
    }
}

使用UDP广播事件

消息POJO：LogEvent

package com.louris.springboot.utils;

import java.net.InetSocketAddress;

public class LogEvent {
    public static final byte SEPARATOR = (byte) ':';
    private final InetSocketAddress source;
    private final String logfile;
    private final String msg;
    private final long received;

    public LogEvent(String logfile, String msg){
        this(null, -1, logfile, msg);
    }

    public LogEvent(InetSocketAddress source, long received,
                    String logfile, String msg){
        this.source = source;
        this.logfile = logfile;
        this.msg = msg;
        this.received = received;
    }

    public InetSocketAddress getSource() {
        return source;
    }

    public long getReceived() {
        return received;
    }

    public String getLogfile() {
        return logfile;
    }

    public String getMsg() {
        return msg;
    }
}

编写广播者

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.socket.DatagramPacket;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.util.CharsetUtil;

import java.net.InetSocketAddress;
import java.util.List;

public class LogEventEncoder extends MessageToMessageEncoder<LogEvent> {
    private final InetSocketAddress remoteAddress;

    public LogEventEncoder(InetSocketAddress remoteAddress){
        this.remoteAddress = remoteAddress;
    }

    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, LogEvent logEvent, List<Object> out) throws Exception {
        byte[] file = logEvent.getLogfile().getBytes(CharsetUtil.UTF_8);
        byte[] msg = logEvent.getMsg().getBytes(CharsetUtil.UTF_8);
        ByteBuf buf = channelHandlerContext.alloc().buffer(file.length + msg.length + 1);
        buf.writeBytes(file);
        buf.writeByte(LogEvent.SEPARATOR);
        buf.writeBytes(msg);
        out.add(new DatagramPacket(buf, remoteAddress));
    }
}

package com.louris.springboot.utils;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;

import java.io.File;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;

public class LogEventBroadcaster {
    private final EventLoopGroup group;
    private final Bootstrap bootstrap;
    private final File file;

    public LogEventBroadcaster(InetSocketAddress address, File file){
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioDatagramChannel.class)
                .option(ChannelOption.SO_BROADCAST, true)
                .handler(new LogEventEncoder(address));
        this.file = file;
    }

    public void run() throws Exception{
        Channel ch = bootstrap.bind(0).sync().channel();
        long pointer = 0;
        for(;;){
            long len = file.length();
            if(len < pointer){
                // file was reset
                pointer = len;
            }else if(len > pointer){
                // Content was added
                RandomAccessFile raf = new RandomAccessFile(file, "r");
                raf.seek(pointer); //设置当前的文件指针，以确保没有任何的旧日志被发送
                String line;
                while((line = raf.readLine()) != null){ //对于每个日志 条目，写入一个LogEvent到Channel中
                    ch.writeAndFlush(new LogEvent(null, -1, file.getAbsolutePath(), line));
                }
                pointer = raf.getFilePointer(); //存储其在文件中的当前位置
                raf.close();
            }
            try{
                Thread.sleep(1000);
            }catch (InterruptedException e){
                Thread.interrupted();
                break;
            }
        }
    }

    public void stop(){
        group.shutdownGracefully();
    }

    public static void main(String[] args) throws Exception{
        if(args.length != 2){
            throw new IllegalArgumentException();
        }
        LogEventBroadcaster broadcaster = new LogEventBroadcaster(
                new InetSocketAddress("255.255.255.255",
                        Integer.parseInt(args[0])), new File(args[1]));
        try{
            broadcaster.run();
        }finally {
            broadcaster.stop();
        }
    }
}

编写监听器

package com.louris.springboot.utils;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.socket.DatagramPacket;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.util.CharsetUtil;

import java.util.List;

public class LogEventDecoder extends MessageToMessageDecoder<DatagramPacket> {

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, DatagramPacket datagramPacket, List<Object> out) throws Exception {
        ByteBuf data = datagramPacket.content();
        int idx = data.indexOf(0, data.readableBytes(), LogEvent.SEPARATOR);
        String filename = data.slice(0, idx).toString(CharsetUtil.UTF_8);
        String logMsg = data.slice(idx + 1, data.readableBytes()).toString(CharsetUtil.UTF_8);
        LogEvent event = new LogEvent(datagramPacket.sender(),
                System.currentTimeMillis(), filename, logMsg);
        out.add(event);
    }
}

package com.louris.springboot.utils;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class LogEventHandler extends SimpleChannelInboundHandler<LogEvent> {

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

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, LogEvent logEvent) throws Exception {
        StringBuilder builder = new StringBuilder();
        builder.append(logEvent.getReceived());
        builder.append(" [");
        builder.append(logEvent.getSource().toString());
        builder.append("] [");
        builder.append(logEvent.getLogfile());
        builder.append("] : ");
        builder.append(logEvent.getMsg());
        System.out.println(builder.toString());
    }
}

package com.louris.springboot.utils;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;

import java.net.InetSocketAddress;

public class LogEventMonitor {
    private final EventLoopGroup group;
    private final Bootstrap bootstrap;

    public LogEventMonitor(InetSocketAddress address){
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioDatagramChannel.class)
                .option(ChannelOption.SO_BROADCAST, true)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel) throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new LogEventDecoder());
                        pipeline.addLast(new LogEventHandler());
                    }
                }).localAddress(address);
    }

    public Channel bind(){
        return bootstrap.bind().syncUninterruptibly().channel(); //注意DatagramChannel是无连接的
    }

    public void stop(){
        group.shutdownGracefully();
    }
    
    public static void main(String[] args) throws Exception{
        if(args.length != 1){
            throw new IllegalArgumentException("Usage: LogEventMonitor <port>");
        }
        LogEventMonitor monitor = new LogEventMonitor(new InetSocketAddress(Integer.parseInt(args[0])));
        try{
            Channel channel = monitor.bind();
            System.out.println("LogEventMonitor running");
            channel.closeFuture().sync();
        }finally {
            monitor.stop();
        }
    }
    
}

Netty——异步和事件驱动

Java网络编程

阻塞I/O

NIO

Netty

异步与事件驱动

Netty的核心组件

Channel

回调

Future

事件和ChannelHandler

Netty应用程序初探

编写Echo服务器

ChannelHandler与业务逻辑

引导服务器

编写Echo客户端

通过ChannelHandler实现客户端逻辑

引导客户端

SpringBoot启动测试

服务器端

客户端

Netty的组件和设计

Channel、EventLoop和ChannelFuture

Channel接口

EventLoop接口

ChannelFuture接口

ChannelHandler和ChannelPipeline

ChannelHandler接口

ChannelPipeline接口

ChannelHandlerAdapter

编码器和解码器

抽象类SimpleChannelInboundHandler

引导

传输

案例研究：传输迁移

传统JDK API的BIO和NIO

BIO

NIO(基于Selector的IO多路复用)

通过Netty使用BIO和NIO

BIO

NIO

传输API

写数据并冲刷到远程节点

内置的传输

NIO——非阻塞I/O

Epoll——用于Linux的本地非阻塞传输

OIO——旧的阻塞I/O

用于JVM内部通信的Local传输

Embedded传输

传输的用例

ByteBuf

ByteBuf的API

ByteBuf类

如何工作

ByteBuf的使用模式

堆缓冲区

直接缓冲区

复合缓冲区

字节级操作

随机访问索引

顺序访问索引

可丢弃字节

可读字节

可写字节

索引管理

查找操作

派生缓冲区

读/写操作

更多的操作

ByteBufHolder接口

ByteBuf分配

按需分配：ByteBufAllocator接口

Unpooled缓冲区

ByteBufUtil类

引用计数

ChannelHandler和ChannelPipeline

ChannelHandlerInboundHandler接口

释放消息资源

资源管理

防止泄漏