Netty 网络编程的秘籍,看这一篇够了

拼搏现实的明天。 2023-10-12 09:35 8阅读 0赞

## Netty ##

> version: 4.1.55.Final

传统的IO模型的web容器，比如老版本的Tomcat，为了增加系统的吞吐量，需要不断增加系统核心线程数量，或者通过水平扩展服务器数量，来增加系统处理请求的能力。 有了NIO之后，一个线程即可处理多个连接事件，基于多路复用模型的Netty框架，不仅降低了使用NIO的复杂度，

### 优点 ###

Netty是一款以java NIO为基础，基于事件驱动模型支持异步、高并发的网络应用框架

*  API使用简单，开发门槛低，简化了NIO开发网络程序的复杂度
 *  功能强大，预置多种编解码功能，支持多种主流协议，比如Http、WebSocket。
 *  定制能力强，可以通过ChannelHandler对通信框架灵活扩展。
 *  性能高，支持异步非阻塞通信模型
 *  成熟稳定，社区活跃，已经修复了Java NIO所有的Bug。
 *  经历了大规模商业应用的考验，质量有保证。

### IO模型 ###

select、poll和epoll

操作系统内核基于这些函数实现非阻塞IO，以此实现多路复用模型

*  select

select

1.  select 调用需要传入 fd 数组，需要拷贝一份到内核，高并发场景下这样的拷贝消耗的资源是惊人的。（可优化为不复制）
2.  select 在内核层仍然是通过遍历的方式检查文件描述符的就绪状态，是个同步过程，只不过无系统调用切换上下文的开销。（内核层可优化为异步事件通知）
3.  select 仅仅返回可读文件描述符的个数，具体哪个可读还是要用户自己遍历。（可优化为只返回给用户就绪的文件描述符，无需用户做无效的遍历）

*  pool

和 select 的主要区别就是，去掉了 select 只能监听 1024 个文件描述符的限制

*  epool

epool

1.  内核中保存一份文件描述符集合，无需用户每次都重新传入，只需告诉内核修改的部分即可。
2.  内核不再通过轮询的方式找到就绪的文件描述符，而是通过异步 IO 事件唤醒。
3.  内核仅会将有 IO 事件的文件描述符返回给用户，用户也无需遍历整个文件描述符集合。

### Reactor模型 ###

一、单Reactor单线程 1）可以实现通过一个阻塞对象监听多个链接请求

2）Reactor对象通过select监听客户端请求事件，通过dispatch进行分发

3）如果是建立链接请求，则由Acceptor通过accept处理链接请求，然后创建一个Handler对象处理完成链接后的各种事件

4）如果不是链接请求，则由Reactor分发调用链接对应的Handler来处理

5）Handler会完成Read->业务处理->send的完整业务流程

reactor

二、单Reactor多线程 1）Reactor对象通过select监听客户端请求事件，收到事件后，通过dispatch分发

2）如果是建立链接请求，则由Acceptor通过accept处理链接请求，然后创建一个Handler对象处理完成链接后的各种事件

3）如果不是链接请求，则由Reactor分发调用链接对应的Handler来处理

4）Handler只负责事件响应不做具体业务处理

5）通过read读取数据后，分发到worker线程池处理，处理完成后返回给Handler，Handler收到后，通过send将结果返回给client

reactor

三、主从Reactor多线程 1）Reactor主线程MainReactor对象通过select监听链接事件，通过Acceptor处理

2）当Acceptor处理链接事件后，MainReactor将链接分配给SubReactor

3）SubReactor将链接加入到队列进行监听，并创建Handler进行事件处理

4）当有新事件发生时，SubReactor就会调用对应的Handler处理

5）Handler通过read读取数据，分发到worker线程池处理，处理完成后返回给Handler，Handler收到后，通过send将结果返回给client

6）Reactor主线程可以对应多个Reactor子线程

reactor

三种模式用生活案例来理解 1）单Reactor单线程，前台接待员和服务员是同一个人，全程为顾客服务

2）单Reactor多线程，1个前台接待员，多个服务员，接待员只负责接待

3）主从Reactor多线程，多个前台接待员，多个服务员

Reactor模型具有如下优点 1）响应快，不必为单个同步事件所阻塞，虽然Reactor本身依然是同步的

2）可以最大程度的避免复杂的多线程及同步问题，并且避免了多线程/进程的切换开销

3）扩展性好，可以方便的通过增加Reactor实例个数来充分利用CPU资源

4）复用性好，Reactor模型本身与具体事件处理逻辑无关，具有很高的复用性

### 核心组件 ###

1.Bootstrap 一个Netty应用通常由一个Bootstrap开始，它主要作用是配置整个Netty程序，串联起各个组件。

Handler，为了支持各种协议和处理数据的方式，便诞生了Handler组件。Handler主要用来处理各种事件，这里的事件很广泛，比如可以是连接、数据接收、异常、数据转换等。

2.ChannelInboundHandler 一个最常用的Handler。这个Handler的作用就是处理接收到数据时的事件，也就是说，我们的业务逻辑一般就是写在这个Handler里面的，ChannelInboundHandler就是用来处理我们的核心业务逻辑。

3.ChannelInitializer 当一个链接建立时，我们需要知道怎么来接收或者发送数据，当然，我们有各种各样的Handler实现来处理它，那么ChannelInitializer便是用来配置这些Handler，它会提供一个ChannelPipeline，并把Handler加入到ChannelPipeline。

4.ChannelPipeline 一个Netty应用基于ChannelPipeline机制，这种机制需要依赖于EventLoop和EventLoopGroup，因为它们三个都和事件或者事件处理相关。

EventLoops的目的是为Channel处理IO操作，一个EventLoop可以为多个Channel服务。

EventLoopGroup会包含多个EventLoop。

5.Channel 代表了一个Socket链接，或者其它和IO操作相关的组件，它和EventLoop一起用来参与IO处理。

6.Future 在Netty中所有的IO操作都是异步的，因此，你不能立刻得知消息是否被正确处理，但是我们可以过一会等它执行完成或者直接注册一个监听，具体的实现就是通过Future和ChannelFutures,他们可以注册一个监听，当操作执行成功或失败时监听会自动触发。

### 示例 ###

通过一个简单的示例，首先了解怎么基于netty开发一个通信程序，包括服务的与客户端：

Server:

@Slf4j
    public class Server {
    
        private EventLoopGroup boosGroup;
    
        private EventLoopGroup workGroup;
    
        public Server(int port){
            try {
                init(port);
                log.info("----- 服务启动成功 -----");
            } catch (InterruptedException e) {
                log.error("启动服务出错：{}", e.getCause());
            }
        }
    
        private void init(int port) throws InterruptedException {
            // 处理连接
            this.boosGroup = new NioEventLoopGroup();
            // 处理业务
            this.workGroup = new NioEventLoopGroup();
    
            ServerBootstrap bootstrap = new ServerBootstrap();
            // 绑定
            bootstrap.group(boosGroup, workGroup)
                    .channel(NioServerSocketChannel.class) //配置服务端
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childOption(ChannelOption.SO_RCVBUF, 1024)
                    .childOption(ChannelOption.SO_SNDBUF, 1024)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new ServerHandler());
                        }
                    });
    
            ChannelFuture channelFuture = bootstrap.bind(port).sync();
            channelFuture.channel().closeFuture().sync();
        }
    
        public void close(){
            this.boosGroup.shutdownGracefully();
            this.workGroup.shutdownGracefully();
        }
    
    }
    
    @Slf4j
    class ServerHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            log.info(">>>>>>> server active");
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            //1. 读取客户端的数据(缓存中去取并打印到控制台)
            ByteBuf buf = (ByteBuf) msg;
            byte[] request = new byte[buf.readableBytes()];
            buf.readBytes(request);
            String requestBody = new String(request, "utf-8");
            log.info(">>>>>>>>> receive message: {}", requestBody);
    
            //2. 返回响应数据
            ctx.writeAndFlush(Unpooled.copiedBuffer((requestBody+" too").getBytes()));
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            super.exceptionCaught(ctx, cause);
        }
    }

Client:

@Slf4j
    public class Client {
    
        private EventLoopGroup workGroup;
        private ChannelFuture channelFuture;
    
        public Client(int port){
            init(port);
        }
    
        private void init(int port){
            this.workGroup = new NioEventLoopGroup();
    
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(workGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                    .option(ChannelOption.SO_RCVBUF, 1024)
                    .option(ChannelOption.SO_SNDBUF, 1024)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new ClientHandler());
                        }
                    });
    
            this.channelFuture = bootstrap.connect("127.0.0.1", port).syncUninterruptibly();
        }
    
        /**
         *
         * @param message
         */
        public void send(String message){
            this.channelFuture.channel().writeAndFlush(Unpooled.copiedBuffer(message.getBytes()));
        }
    
        /**
         *
         */
        public void close(){
            try {
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            workGroup.shutdownGracefully();
        }
    }
    
    @Slf4j
    class ClientHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            log.info(">>>>>>> client active");
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            try {
                ByteBuf buf = (ByteBuf) msg;
                byte[] req = new byte[buf.readableBytes()];
                buf.readBytes(req);
    
                String body = new String(req, "utf-8");
                log.info(">>>>>>>>> receive message: {}", body);
            } finally {
                ReferenceCountUtil.release(msg);
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            super.exceptionCaught(ctx, cause);
        }
    }

测试：

public class StarterTests {
    
        static int port = 9011;
    
        @Test
        public void startServer(){
            Server server = new Server(9011);
        }
    
        @Test
        public void startClient(){
            Client client = new Client(port);
            client.send("Hello Netty!");
            while (true){}
        }
    
    }

### 生态 ###

*  Dubbo
 *  Spring Reactive

### 类似技术 ###

Mina、Netty、Grizzly

### 其他 ###

Proactor非阻塞异步网络模型