网络开发的最强大框架：Netty快速入门

Posted 2021-09-14 Java鱼仔

本文收录于JavaStarter ，里面有我完整的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) {
        // 1、服务器端的启动器，组装netty组件
        new ServerBootstrap()
                //2、组装eventLoop组
                .group(new NioEventLoopGroup())
                //3、选择服务器的ServerSocketChannel实现
                .channel(NioserverSocketChannel.class)
                //4、负责处理连接和读写
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        //将bytebuffer转换为字符串
                        nioSocketChannel.pipeline().addLast(new StringDecoder());
                        //自定义handler，这里接收读事件后展示数据
                        nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                System.out.println(msg);
                            }
                        });
                    }
                })
                //5、监听端口
                .bind(8080);
    }
}

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

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

2、组装eventLoopGroup

3、组装Channel

4、添加handler处理器

5、建立连接

6、发送数据到服务端

public class FirstClient {
    public static void main(String[] args) throws InterruptedException {
        //1、启动类
        new Bootstrap()
                //2、添加EventLoop组
                .group(new NioEventLoopGroup())
                //3、添加channel
                .channel(NioSocketChannel.class)
                //4、添加处理器
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        //将发送的内容encode编码
                        nioSocketChannel.pipeline().addLast(new StringEncoder());
                    }
                })
                //5、连接到服务器
                .connect(new InetSocketAddress("localhost",8080))
                .sync()
                .channel()
                //6、发送数据
                .writeAndFlush("hello");
    }
}

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

（三）理解Netty中的组件

3.1 EventLoop

EventLoop其实是一个单线程的执行器，同时维护了一个Selector，EventLoop的作用是处理Channel上的io事件。

3.2 EventLoopGroup

EventLoopGroup是一组EventLoop，Channel通常会调用EventLoopGroup中的register方法绑定其中的一个EventLoop，后续这个channel中的io事件则都由这个EventLoop处理。

3.3 channel

channel是一个数据的传输流，channel可以理解为是通讯的载体。

3.4 ChannelHandler

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

3.5 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

5.1 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");
        }
    });
}

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

5.2 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更加容易理解一些。我是鱼仔，我们下期再见！