MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Netty核心组件详解及应用

2021-07-267.8k 阅读

Netty 核心组件之 Channel

在 Netty 的世界里，Channel 是网络操作的基础抽象，它代表了到网络套接字或者连接的一个通道。它类似于 Java NIO 中的 SelectableChannel，但提供了更丰富的功能和更友好的使用方式。

Channel 接口定义了许多操作，包括绑定、连接、读写等。例如，bind(SocketAddress local) 方法用于将 Channel 绑定到本地地址，connect(SocketAddress remote) 方法用于连接到远程地址。

下面是一个简单的 NioSocketChannel 创建示例代码：

EventLoopGroup group = new NioEventLoopGroup();
try {
    Bootstrap b = new Bootstrap();
    b.group(group)
      .channel(NioSocketChannel.class)
      .handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new EchoClientHandler());
            }
        });

    ChannelFuture f = b.connect(new InetSocketAddress("127.0.0.1", 8080)).sync();
    f.channel().closeFuture().sync();
} catch (InterruptedException e) {
    e.printStackTrace();
} finally {
    group.shutdownGracefully();
}

在上述代码中，NioSocketChannel 是 Channel 的一个具体实现，用于基于 NIO 的客户端套接字通道。通过 Bootstrap 来配置并创建 NioSocketChannel 实例，并尝试连接到指定的服务器地址。

ChannelPipeline 和 ChannelHandler

ChannelPipeline 是 Channel 所有处理逻辑的容器，它由一系列的 ChannelHandler 组成，这些 ChannelHandler 负责处理入站和出站数据。可以把 ChannelPipeline 想象成一条流水线，数据从一端进入，经过一系列的处理步骤后从另一端流出。

ChannelHandler 分为两种类型：入站处理器（ChannelInboundHandler）和出站处理器（ChannelOutboundHandler）。入站处理器负责处理从远程节点接收的数据，而出站处理器负责处理发送到远程节点的数据。

自定义 ChannelHandler 示例

public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        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 {
        ctx.flush();
    }

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

上述代码定义了一个简单的入站处理器 EchoServerHandler，它继承自 ChannelInboundHandlerAdapter。在 channelRead 方法中，它读取从客户端发送过来的数据并打印，然后将数据写回给客户端。channelReadComplete 方法在数据读取完成后被调用，用于刷新缓冲区确保数据被发送出去。exceptionCaught 方法用于处理在处理过程中发生的异常。

要将 ChannelHandler 添加到 ChannelPipeline 中，可以使用如下代码：

ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
  .channel(NioServerSocketChannel.class)
  .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new EchoServerHandler());
        }
    });

在这个服务器端启动代码中，通过 ChannelInitializer 的 initChannel 方法，将 EchoServerHandler 添加到了 ChannelPipeline 中。

EventLoop 和 EventLoopGroup

EventLoop 是 Netty 处理 I/O 操作的核心组件，它负责处理注册到它上面的 Channel 的 I/O 事件。每个 EventLoop 会不断循环处理它所负责的 Channel 的 I/O 事件队列。

EventLoopGroup 是一组 EventLoop 的抽象，它提供了一种管理多个 EventLoop 的方式。在服务器端编程中，通常会使用两个 EventLoopGroup，一个用于接受客户端连接（通常称为 bossGroup），另一个用于处理已连接客户端的 I/O 操作（通常称为 workerGroup）。

创建 EventLoopGroup 示例

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
    ServerBootstrap b = new ServerBootstrap();
    b.group(bossGroup, workerGroup)
      .channel(NioServerSocketChannel.class)
      .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new EchoServerHandler());
            }
        });

    ChannelFuture f = b.bind(new InetSocketAddress(8080)).sync();
    f.channel().closeFuture().sync();
} catch (InterruptedException e) {
    e.printStackTrace();
} finally {
    bossGroup.shutdownGracefully();
    workerGroup.shutdownGracefully();
}

在上述代码中，创建了一个单线程的 bossGroup 用于接受连接，和一个默认线程数的 workerGroup 用于处理 I/O 操作。通过 ServerBootstrap 的 group 方法将这两个 EventLoopGroup 应用到服务器启动配置中。

ByteBuf

ByteBuf 是 Netty 提供的一个高效的字节缓冲区，用于替代 Java NIO 中的 ByteBuffer。ByteBuf 提供了更灵活和高效的读写操作。

ByteBuf 的读写操作

ByteBuf buf = Unpooled.buffer(10);
buf.writeByte(1);
buf.writeShort(2);
buf.writeInt(3);

byte b = buf.readByte();
short s = buf.readShort();
int i = buf.readInt();

上述代码首先创建了一个容量为 10 的 ByteBuf，然后依次写入一个字节、一个短整型和一个整型数据。之后从 ByteBuf 中按顺序读取相应的数据。

ByteBuf 有两种类型的索引：读索引（readerIndex）和写索引（writerIndex）。每次读取操作会增加 readerIndex，每次写入操作会增加 writerIndex。

ByteBuf 的内存管理

Netty 提供了两种主要的 ByteBuf 内存分配方式：堆内存（HeapByteBuf）和直接内存（DirectByteBuf）。堆内存分配和回收速度快，但在进行 I/O 操作时需要额外的拷贝；直接内存分配和回收速度相对较慢，但在 I/O 操作时可以避免额外的拷贝，提高性能。

ByteBuf heapBuf = UnpooledHeapByteBufAllocator.DEFAULT.buffer(10);
ByteBuf directBuf = UnpooledDirectByteBufAllocator.DEFAULT.buffer(10);

上述代码分别创建了堆内存和直接内存的 ByteBuf。在实际应用中，需要根据具体的场景来选择合适的内存分配方式。

Bootstrap 和 ServerBootstrap

Bootstrap 用于客户端或者无连接协议（如 UDP）的服务端配置和启动。ServerBootstrap 则专门用于 TCP 服务端的配置和启动。

客户端 Bootstrap 示例

EventLoopGroup group = new NioEventLoopGroup();
try {
    Bootstrap b = new Bootstrap();
    b.group(group)
      .channel(NioSocketChannel.class)
      .handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new EchoClientHandler());
            }
        });

    ChannelFuture f = b.connect(new InetSocketAddress("127.0.0.1", 8080)).sync();
    f.channel().closeFuture().sync();
} catch (InterruptedException e) {
    e.printStackTrace();
} finally {
    group.shutdownGracefully();
}

在这个客户端示例中，通过 Bootstrap 配置了 NioSocketChannel 作为客户端通道，并添加了自定义的 EchoClientHandler 到 ChannelPipeline 中，最后连接到指定的服务器地址。

服务器端 ServerBootstrap 示例

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
    ServerBootstrap b = new ServerBootstrap();
    b.group(bossGroup, workerGroup)
      .channel(NioServerSocketChannel.class)
      .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new EchoServerHandler());
            }
        });

    ChannelFuture f = b.bind(new InetSocketAddress(8080)).sync();
    f.channel().closeFuture().sync();
} catch (InterruptedException e) {
    e.printStackTrace();
} finally {
    bossGroup.shutdownGracefully();
    workerGroup.shutdownGracefully();
}

在这个服务器端示例中，ServerBootstrap 使用两个 EventLoopGroup，bossGroup 用于接受连接，workerGroup 用于处理已连接客户端的 I/O 操作。通过 childHandler 方法为每个新连接的客户端添加 EchoServerHandler 到 ChannelPipeline 中，并绑定到指定的端口。

编解码器（Codec）

在网络通信中，数据需要进行编码和解码操作。Netty 提供了丰富的编解码器实现，方便开发者处理不同协议的数据编解码。

字符串编解码器示例

public class StringServer {
    private static final int PORT = 8080;

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
              .channel(NioServerSocketChannel.class)
              .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
                        ch.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
                        ch.pipeline().addLast(new StringServerHandler());
                    }
                });

            ChannelFuture f = b.bind(PORT).sync();
            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

public class StringServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String request = (String) msg;
        System.out.println("Server received: " + request);
        String response = "Response to: " + request;
        ctx.writeAndFlush(response);
    }

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

在上述代码中，通过 StringDecoder 和 StringEncoder 实现了字符串的编解码。StringDecoder 将接收到的字节数据解码为字符串，StringEncoder 将字符串编码为字节数据以便发送。StringServerHandler 处理解码后的字符串数据，并返回响应。

自定义编解码器

有时候，需要根据特定的协议来实现自定义的编解码器。假设我们有一个简单的协议，消息格式为：4 字节长度 + 消息内容。

public class CustomProtocolDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        if (in.readableBytes() < 4) {
            return;
        }
        in.markReaderIndex();
        int length = in.readInt();
        if (in.readableBytes() < length) {
            in.resetReaderIndex();
            return;
        }
        ByteBuf msg = in.readBytes(length);
        out.add(msg);
    }
}

public class CustomProtocolEncoder extends MessageToByteEncoder<ByteBuf> {
    @Override
    protected void encode(ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception {
        out.writeInt(msg.readableBytes());
        out.writeBytes(msg);
    }
}

上述代码定义了 CustomProtocolDecoder 用于将接收到的字节数据解码为符合协议格式的消息，CustomProtocolEncoder 用于将消息编码为符合协议格式的字节数据。在使用时，将这两个编解码器添加到 ChannelPipeline 中即可。

Netty 的应用场景

Netty 在许多领域都有广泛的应用，以下是一些常见的场景：

高性能网络服务器

Netty 凭借其高效的 I/O 处理和灵活的架构，非常适合构建高性能的网络服务器，如游戏服务器、即时通讯服务器等。例如，许多大型游戏的后端服务器使用 Netty 来处理大量客户端的连接和实时数据交互。

分布式系统

在分布式系统中，Netty 可以用于实现节点之间的高效通信。通过 Netty 可以构建可靠的远程调用框架，实现分布式服务之间的通信。例如，一些微服务框架会基于 Netty 来实现服务间的通信协议。

物联网（IoT）

在物联网场景中，设备之间需要进行高效、可靠的通信。Netty 可以处理大量设备的连接，并对不同协议的数据进行编解码。例如，智能家居系统中的设备与服务器之间的通信可以使用 Netty 来实现。

总结

Netty 的核心组件包括 Channel、ChannelPipeline、ChannelHandler、EventLoop、EventLoopGroup、ByteBuf、Bootstrap 和 ServerBootstrap 以及编解码器等。这些组件相互协作，为开发者提供了一个高效、灵活的网络编程框架。通过合理地使用这些组件，可以构建出高性能、可扩展的网络应用程序，应用于各种不同的领域。无论是开发高性能网络服务器、分布式系统还是物联网应用，Netty 都能发挥其强大的功能。在实际开发中，需要根据具体的需求和场景，深入理解并合理运用这些核心组件，以实现最优的性能和功能。同时，通过不断地实践和优化，能够更好地掌握 Netty 的使用技巧，开发出更加健壮和高效的网络应用。