作者:鱼仔
博客首页: codeease.top (opens new window)
公众号:Java鱼仔

什么是netty

Netty 是一个异步的，基于事件驱动的网络应用框架，用于快速开发可维护、高性能的网络服务器和客户端。Netty 的应用十分广泛，可以说主流的框架中，如果有网络方面的需求，一般用的都是 Netty 框架。比如 Dubbo、ES、**Zookeeper **中都用到了 Netty。因此即使在平常工作中没有 Netty 的使用场景，Netty 还是十分值得我们去学习的。

Netty 底层基于 NIO 开发，其实大部分的 Java 程序员对于网络方面的开发能力是比较弱的，因此如果有网络相关的开发业务，如果自己通过 BIO 或者 NIO 实现，会产生很多问题。而通过 Netty 可以快速开发网络应用，因此也有人把 Netty 称为网络开发框架中的 Spring。

关于 NIO 和 BIO 的区别，我之前在博客中也讲到过，BIO 每次通信都要新建一个线程去处理，NIO 通过多路复用的方式去处理请求。下图就是 NIO 的处理流程。

第一个 Netty 入门程序

既然基于 NIO 开发，Netty 的入门程序和我们当时写的 NIO 入门程序比较像，首先开发一个服务器端，Netty 的开发流程可以遵循一套规范：

1. 通过 ServerBootstrap 启动，组装 Netty 组件
2. 组装 EventLoopGroup
3. 组装 Channel
4. 通过 Handler 处理连接、读写请求

public class FirstServer {
    public static void main(String[] args) {
        // 创建两个NioEventLoopGroup实例，分别用于处理boss（接受连接）和worker（处理连接）任务
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();

        try {
            // 创建一个新的ServerBootstrap实例，用于配置服务器
            ChannelFuture future = new ServerBootstrap()
                    // 指定bossGroup和workGroup，分别用于接受连接和处理连接
                    .group(bossGroup, workGroup)
                    // 指定使用的通道类型为NioServerSocketChannel，适用于NIO传输
                    .channel(NioServerSocketChannel.class)
                    // 设置子通道的处理器
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // 获取通道的pipeline，用于添加处理器
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            // 添加一个字符串解码器，将ByteBuf转换为字符串
                            pipeline.addLast(new StringDecoder());
                            // 添加一个简单的通道处理器，处理接收到的字符串消息
                            pipeline.addLast(new SimpleChannelInboundHandler<String>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
                                    // 打印接收到的消息
                                    System.out.println(msg);
                                }
                            });
                        }
                    }).bind(8080).sync();
            // 等待服务器socket通道关闭
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            // 处理可能的异常
            e.printStackTrace();
        } finally {
            // 优雅地关闭bossGroup和workGroup，释放资源
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

接着开发一个客户端，客户端的整体流程和服务器端十分类似：

1、通过 Bootstrap 启动，组装 Netty 组件

2、组装 EventLoopGroup

3、组装 Channel

4、添加 Handler 处理器

5、建立连接

6、发送数据到服务端

public class FirstClient {
    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap()
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    System.out.println("client " + ctx);
                                    ctx.writeAndFlush(Unpooled.copiedBuffer("客户端发了一条消息", CharsetUtil.UTF_8));
                                }
                                
                                @Override
                                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                                    cause.printStackTrace();
                                    ctx.close();
                                }
                            });
                        }
                    });
            System.out.println("客户端 ok..");
            //启动客户端去连接服务器端
            //关于 ChannelFuture 要分析，涉及到netty的异步模型
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
            //给关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            group.shutdownGracefully();
        }

    }
}

启动服务器端后再启动客户端，可以发现服务端接受到了客户端发过来的信息。看到这里觉得还是有点蒙没关系，下面会对每个组件进行讲解。

理解Netty中的组件

EventLoop

EventLoop 其实是一个单线程的执行器，同时维护了一个 Selector，EventLoop 的主要任务是处理网络事件，包括 IO 事件、定时任务和用户自定义任务。

EventLoopGroup

EventLoopGroup 是一组 EventLoop 的集合，管理一组 EventLoop 实例。它负责分配和管理这些事件循环线程。

Netty 服务端通常分为两组线程池（事件组）来处理不同类型的任务，BossEventLoopGroup 和 WorkerEventLoopGroup。

BossEventLoopGroup 负责监听 ServerSocketChannel 的连接事件，接收到新连接后，将其注册到 WorkerEventGroup。

WorkerEventLoopGroup 负责处理与已建立连接相关的所有 IO 事件（如读写操作）。

Bootstrap

Bootstrap 的作用是串联整个 Netty 程序，同时他也是 Netty 程序的入口。ServerBootstrap 是 Netty 服务端的入口，Bootstrap 是 Netty 客户端的入口。

Channel

Channel 是一个数据的传输流，Channel 可以理解为是通讯的载体。Channel可以提供下面一系列功能

网络I/O操作

• 读取数据：从远程端接收数据。
• 写入数据：将数据发送到远程端。

管理连接状态

• 检测连接是否打开、是否激活等。
• 提供关闭连接的操作。

关联事件循环

• 每个 Channel 与一个 EventLoop 绑定，用于处理 I/O 操作和事件通知。

支持异步操作

• Netty 的 I/O 操作（如写入或关闭连接）通常是异步的，Channel 提供了 ChannelFuture 接口，用于监听操作完成的结果。

常用的Channel

• NioSocketChannel，异步的客户端 TCP Socket 连接。
• NioServerSocketChannel，异步的服务器端 TCP Socket 连接。
• NioDatagramChannel，异步的 UDP 连接。

ChannelHandler

ChannelHandler 是用来处理 Channel 上的各种事件的，所有的 ChannelHandler 连起来就是 pipeline。简单来讲，Channel 是数据的传输通道，而 ChannelHandler 用来处理通道中的数据。

ByteBuf

ByteBuf 是 netty 中数据的传输载体，网络数据的基本单位总是字节，ByteBuf 用来传输这些网络上的字节。

一、EventLoop

EventLoop可以处理多种任务，单独使用EventLoop可以通过下面几个步骤实现：

1、创建一个EventLoopGroup

2、从EventLoopGroup中获取EventLoop

3、通过EventLoop执行任务

通过代码这样表示：

public class TestEventLoop {
    public static void main(String[] args) {
        //1、创建事件循环组
        //NioEventLoopGroup可以处理IO事件、普通任务、定时任务
        EventLoopGroup group=new NioEventLoopGroup();
        //2、获取下一个事件循环对象
        EventLoop eventLoop = group.next();
        //3、执行普通任务
        eventLoop.execute(()->{
            System.out.println("普通任务");
        });
        //4、执行定时任务
        eventLoop.scheduleAtFixedRate(()->{
            System.out.println("定时任务");
        },0,1, TimeUnit.SECONDS);
    }
}

EventLoop最常用的就是执行IO任务了，我们在入门程序中写的group(new NioEventLoopGroup())就是把EventLoop用来处理IO任务。

在netty中，我们还会在绑定group时指定boss和work，boss用来处理连接，work用来处理收到读写请求后续的操作，有的时候我们还可以自定义EventLoopGroup处理其他任务，因此前面的FirstServer 可以写成下面这样：

public class NioServer {
    public static void main(String[] args) {
        //boss用来处理连接
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        //work用来处理读写请求
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        //otherGroup处理普通任务，比如打印一段内容
        EventLoopGroup otherGroup=new DefaultEventLoop();
        
        new ServerBootstrap()
                .group(bossGroup,workGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf byteBuf = (ByteBuf) msg;
                                System.out.println(byteBuf.toString());
                                ctx.fireChannelRead(msg); //将msg传给下一个处理者
                            }
                        })
                        .addLast(otherGroup,"handler",new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf byteBuf = (ByteBuf) msg;
                                System.out.println(byteBuf.toString());
                            }
                        });
                    }
                }).bind(8080);
    }
}

二、Channel

Channel 中有几个常用的方法：

close() 关闭channel
pipeline() 添加处理器
write() 将数据写入到缓冲区
flush() 将数据刷出，也就是发给服务端
writeAndFlush() 将数据写入并刷出

我们通过入门案例的客户端代码讲解 Channel

Channel 的连接

public class NioClient  {
    public static void main(String[] args) throws InterruptedException {
        ChannelFuture channelFuture = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        nioSocketChannel.pipeline().addLast(new StringEncoder());
                    }
                })
                //connect是一个异步调用的过程，因此必须要使用sync方法等待连接建立
                .connect(new InetSocketAddress("localhost", 8080));
        //1、使用sync方法阻塞线程直到连接建立
        channelFuture.sync();
        channelFuture.channel().writeAndFlush("hello,world");
    }
}

整个流程这里就不介绍了，主要介绍里面的一个方法 channelFuture.sync();
当调用 connect 方法建立连接时，这个 connect 方法其实是一个异步的方法，因此如果不加 channelFuture.sync() 方法等待连接建立，是无法获取到连接后的 channel 的，更别提写入数据了。

除了使用sync等待连接，还可以采用设置监听器的方式获取 channelFuture

public static void main(String[] args) throws InterruptedException {
    ChannelFuture channelFuture = new Bootstrap()
            .group(new NioEventLoopGroup())
            .channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                    nioSocketChannel.pipeline().addLast(new StringEncoder());
                }
            })
            //connect是一个异步调用的过程，因此必须要使用sync方法等待连接建立
            .connect(new InetSocketAddress("localhost", 8080));
    //2、使用addListener方法异步处理结果
    channelFuture.addListener(new ChannelFutureListener() {
        //在nio连接建立完毕之后，调用operationComplete方法
        @Override
        public void operationComplete(ChannelFuture channelFuture) throws Exception {
            Channel channel = channelFuture.channel();
            channel.writeAndFlush("hello,world");
        }
    });
}

思路是一样的，等连接建立之后再处理对应的方法。

Channel 的关闭

除了连接是异步方法之外，Channel 的关闭方法也是异步的，因此也需要通过

同步阻塞的方式等待关闭：
Channel channel = channelFuture.channel();
ChannelFuture closeFuture = channel.closeFuture();
System.out.println("等待关闭中");
//当其他线程关闭了channel,sync同步等待
closeFuture.sync();
System.out.println("连接已关闭");

同样也可以采用监听器回调的方式：

Channel channel = channelFuture.channel();
ChannelFuture closeFuture = channel.closeFuture();
closeFuture.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        System.out.println("连接已关闭");
    }
});

三、ChannelHandler

ChannelHandler 是用来处理 Channel 上的各种事件的，handler 分为 **inbound **和 **outbount **两种，所有的ChannelHandler 连起来就是 pipeline。

**ChannelInboundHandlerAdapter **的子类主要用来读取客户端数据，写回结果。

**ChannelOutboundHandlerAdapter **的字类主要对写回结果进行加工。

关于 handler 和 pipeline 的代码在前面的例子中都有写

public static void main(String[] args) {
    new ServerBootstrap()
            .group(new NioEventLoopGroup())
            .channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                     //1、从channel中获取到pipeline
                     ChannelPipeline pipeline = nioSocketChannel.pipeline();
                     //2、添加handler处理器到pipeline
                    pipeline.addLast(new ChannelInboundHandlerAdapter(){
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            ByteBuf byteBuf = (ByteBuf) msg;
                            System.out.println(byteBuf.toString());
                            ctx.fireChannelRead(msg); //将msg传给下一个处理者
                        }
                    });
                    //3、添加多个表示依次执行
                    pipeline.addLast(new ChannelInboundHandlerAdapter(){
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            ByteBuf byteBuf = (ByteBuf) msg;
                            System.out.println(byteBuf.toString());
                        }
                    });
                }
            }).bind(8080);

四、ByteBuf

netty中的ByteBuf比JDK自带的ByteBuffer对字节数据的操作更加友好，也更加强大。ByteBuf的主要优势有几下几点：

1、支持自动扩容

2、支持池化技术，可以重用实例，节约内存

3、读写指针分离

4、很多方法体现了零拷贝，比如slice、duplicate等

接下来通过一些操作带你来了解ByteBuf。

1、自动扩容

创建一个默认的ByteBuf，初始容量是256，写入一系列数据之后，这个容量会随着数据的增大自动扩容。

public static void main(String[] args) {
    ByteBuf buf= ByteBufAllocator.DEFAULT.buffer();
    System.out.println(buf);
    StringBuilder stringBuilder=new StringBuilder();
    for (int i = 0; i < 500; i++) {
        stringBuilder.append("1");
    }
    buf.writeBytes(stringBuilder.toString().getBytes());
    System.out.println(buf);
}

结果：

PooledUnsafeDirectByteBuf(ridx: 0, widx: 0, cap: 256)
PooledUnsafeDirectByteBuf(ridx: 0, widx: 500, cap: 512)

ByteBuf的扩容规则如下：

如果数据大小没有超过512，每次扩容到16的整数倍

如果数据大小超过512，则扩容到下一个2^n次

扩容不能超过max capacity

2、直接内存和堆内存

ByteBuf支持创建基于直接内存的ByteBuf，也支持创建基于堆内存的ByteBuf。两者的差距在于：

堆内存的分配效率较高，但是读写性能相对比较低。

直接内存的分配效率比较低，但是读写性能较高（少一次内存复制）

netty默认使用直接内存作为创建ByteBuf的方式

ByteBufAllocator.DEFAULT.heapBuffer();
ByteBufAllocator.DEFAULT.directBuffer();

3、池化技术

ByteBuf支持池化技术，所谓池化指的是ByteBuf创建出来后可以重用，节约内存。通过JVM参数开启或关闭netty的池化，默认开启状态：

-Dio.netty.allocator.type={unpooled|pooled}

总结

Netty是个很强大的框架，但是网络开发本就是一件比较复杂的事情，接下来我会用Netty做一些简单的应用出来，通过这些应用会让Netty更加容易理解一些。我是鱼仔，我们下期再见！