netty代码流程+源码解析

小鱼儿 2023-02-23 08:12 23阅读 0赞

# netty连接 #

p
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.PooledByteBufAllocator;
    
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    import javax.annotation.PreDestroy;
    
    @Component
    public class ModbusTcpSlave { 
    
        private  final HikGaLogger LOGGER = HikGaLoggerFactory.getLogger(ModbusTcpSlave.class);
    
    
    
        @Value("${spring.server.ip}")
        private String localIp;
    
        @Value("${modbus.server.port}")
        private  int modbusServerPort;
        
        @Autowired
        @SuppressWarnings("all")
        private ModbusBeanMap modBusBeanMap;
    
        /* netty的整体流程： 1.初始化创建 2 个 NioEventLoopGroup：其中 boosGroup 用于Accetpt 连接建立事件并分发请求，workerGroup 用于处理 I/O读写事件和业务逻辑。 2.基于 ServerBootstrap(服务端启动引导类)：配置 EventLoopGroup、Channel 类型，连接参数、配置入站、出站事件 handler。 3.绑定端口：开始工作。 */
        EventLoopGroup bossGroup; //用于处理客户端的连接请求
        EventLoopGroup workerGroup; //处理与各个客户端连接的 IO 操作。
    
        @PostConstruct
        public void init() { 
    
            try { 
                // 创建mainReactor
                bossGroup = new NioEventLoopGroup();
                // 创建工作线程组
                workerGroup = new NioEventLoopGroup();
    
                //Bootstrap 意思是引导，一个 Netty 应用通常由一个 Bootstrap 开始，主要作用是配置整个 Netty 程序，串联各个组件，
                // Netty 中 Bootstrap 类是客户端程序的启动引导类，ServerBootstrap 是服务端启动引导类。
                ServerBootstrap bootstrap = new ServerBootstrap();
    
                // 向pipeline中添加编码、解码、业务处理的handler
                ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() { 
                    /** * * @param channel * @throws Exception */
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception { 
                        channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 4,
                                2, 0,0,false)); //处理粘包
                        channel.pipeline().addLast(new ModbusTcpDecoder(modBusBeanMap));
                        channel.pipeline().addLast(new ModbusTcpHandler(modBusBeanMap));
                        channel.pipeline().addLast(new ModbusTcpEncoder(modBusBeanMap));
                    }
                };
    
                // 组装NioEventLoopGroup
                bootstrap.group(bossGroup, workerGroup)
                        // 指定通道channel的类型处理连接请求，由于是服务端，故而是NioServerSocketChannel。实际上在设置channelFactory
                        .channel(NioServerSocketChannel.class)
                        // 设置子通道也就是SocketChannel的处理器， 其内部是实际业务开发的"主战场"
                        .childHandler(initializer)
                        // 设置tcp协议的请求等待队列。。设置连接配置参数
                        .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
                 /* * * ChannelOption.SO_KEEPALIVE, true * 是否启用心跳保活机制。在双方TCP套接字建立连接后（即都进入ESTABLISHED状态）并且在两个小时左右 * 上层没有任何数据传输的情况下，这套机制才会被激活。 */
                // .childOption(ChannelOption.SO_KEEPALIVE, true);
                //Netty中的I/O 操作是异步的，包括 Bind、Write、Connect 等操作会简单的返回一个 ChannelFuture。
                //当 Future 对象刚刚创建时，处于非完成状态，调用者可以通过返回的ChannelFuture 来获取操作执行的状态，注册监听函数来执行完成后的操作
                //绑定并侦听某个端口
                //.bind()绑定端口，该方法是异步执行，所以需要配置监听器
                ChannelFuture future = bootstrap.bind( "127.0.0.1",17002).sync();
                if (future.isSuccess()) { 
                    LOGGER.info("初始化成功");
                } else { 
                    LOGGER.error("初始化异常：", future.cause());
                }
    // ChannelFuture f = bootstrap.bind( "10.192.138.21",17009).addListener((future) -> { 
    // if (future.isSuccess()) { 
    // LOGGER.info("modbus service initialized successfully");
    // } else { 
    // LOGGER.errorWithErrorCode(ErrorCode.MODBUS_SERVICE_INITIALIZED_EXCEPTION.getCode(),
    // ErrorCode.MODBUS_SERVICE_INITIALIZED_EXCEPTION.getMessage());
    // }
    // }).sync(); //直接阻塞，不可用！！！
    
                //程序执行到这里开始阻塞,等待有计算机连接进来
               // future.channel().closeFuture().sync();
    
            } catch (Exception e){ 
                throw new BusinessException(ErrorCode.MODBUS_SERVICE_INITIALIZED_EXCEPTION.getCode(),
                        ErrorCode.MODBUS_SERVICE_INITIALIZED_EXCEPTION.getMessage(), e);
            }
        }
    
        @PreDestroy
        public void destroy() { 
            if (workerGroup != null) { 
                workerGroup.shutdownGracefully();
            }
            if (bossGroup != null) { 
                bossGroup.shutdownGracefully();
            }
        }
    
    }

在bootstrap.bind( localIp,17009)方法中，通常只绑定端口，ip默认为本机ip，无需特殊绑定。

### 具体步骤见下： ###

（1）new了两个NioEventLoopGroup，一个是boss，一个是work。NioEventLoopGroup是一个线程组，boss线程组用于接收客户端的连接工作，work线程组用于处理数据。  
（2）在start方法中，首先new了一个ServerBootstrap，，名字叫bootstrap。它是netty用于启动NIO服务端的辅助启动类。目的是降低服务端的开发复杂度。  
（3）.group(boss, work)目的是将两个线程组传入，让其工作。  
（4）.channel(NioServerSocketChannel.class)就类比与NIO中的ServerSocketChannel。  
（5）.option(ChannelOption.SO\_BACKLOG, 1024)配置TCP参数，将其中一个参数backlog设置为1024，表明临时存放已完成三次握手的请求的队列的最大长度。  
（6）.childOption(ChannelOption.SO\_KEEPALIVE, true)设置TCP长连接,一般如果两个小时内没有数据的通信时,TCP会自动发送一个活动探测数据报文。  
（7）.handler(new LoggingHandler(LogLevel.INFO))处理Log日志。  
（8）.childHandler，用于处理客户端的IO事件，比如有一个客户端发起请求，要读取数据，就可以使用这里面的类来处理这个事件。这是整个处理的核心。也是我们自己主要关注的类。  
（9）ChannelFuture cf = bootstrap.bind(8888).sync();绑定监听端口。***使用sync方法阻塞一直到绑定成功***（等待服务器启动完毕，才会进入下行代码）。  
（10）下面通过一个if语句表明，如果绑定成功那就输出“服务端启动成功”。  
（11）最后bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); 关闭两组死循环。优雅退出~~

### 关于Bootstrap的handler和childHandler区别 ###

handler()和childHandler()的主要区别是，handler()是发生在初始化的时候，childHandler()是发生在客户端连接之后。

也就是说，如果需要在客户端连接前的请求进行handler处理，则需要配置handler(),如果是处理客户端连接之后的handler,则需要配置在childHandler()。

其实，option和childOption也是一样的道理。

### 关于bootstrap.bind( “127.0.0.1”,17002).sync()； ###

> Q1 为什么绑定端口要设置成异步的？又不会阻塞？\*\*  
> 答：bootstrap.bind(port).sync();这个代码底层就是server.accpet()建立连接。bootstrap.bind(port).sync()对其进行了封装，先绑定端口号再建立连接，server.accpet()；这个是阻塞的，所以是异步；

> Q2：bootstrap.bind(port)是绑定端口 .channel(NioServerSocketChannel.class)是负责监听的，为什么是先监听代码后绑定代码呢
> 
> 答：不是的，.channel(NioServerSocketChannel.class)是通道标识，这个代码是客户端还是服务器

### 关于future.channel().closeFuture().sync(); ###

等待服务器 socket 关闭 在这个例子中，这不会发生，但你可以优雅地关闭你的服务器。

// 组装NioEventLoopGroup
                bootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(initializer)
                        .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    
                ChannelFuture future = bootstrap.bind( "10.192.138.21",17002).sync();
                if (future.isSuccess()) { 
                    LOGGER.info("初始化成功");
                } else { 
                    LOGGER.error("初始化异常：", future.cause());
                }
    
                future.channel().closeFuture().sync();
            } catch (Exception e){ 
                throw new BusinessException(ErrorCode.MODBUS_SERVICE_INITIALIZED_EXCEPTION.getCode(),
                        ErrorCode.MODBUS_SERVICE_INITIALIZED_EXCEPTION.getMessage(), e);
            }finally { 
                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
        }

**当future.channel().closeFuture().sync();放出来时**，线程会一直wait，不会执行finally的语句，之后的bean也实例化不成功。（***导致springboot主线程阻塞，无法继续加载剩下的bean***）--------让线程进入wait状态，也就是main线程暂时不会执行到finally里面，nettyserver也持续运行，如果监听到关闭事件，可以优雅的关闭通道和nettyserver，虽然这个例子中，永远不会监听到关闭事件。也就是说这个例子是仅仅为了展示存在api shutdownGracefully，可以优雅的关闭nettyserver。  
**当future.channel().closeFuture().sync();注释掉时**，服务启动时会执行finally的关闭服务器的方法shutdownGracefully，此时客户端时无法连接上server端，因为服务线程已经被关闭了。

**正确是方式是：**

ChannelFuture future = bootstrap.bind( "127.0.0.1",17002).sync();
                if (future.isSuccess()) { 
                    LOGGER.info("初始化成功");
                } else { 
                    LOGGER.error("初始化异常：", future.cause());
                }
    
                //future.channel().closeFuture().sync();
            } catch (Exception e){ 
                throw new BusinessException(ErrorCode.MODBUS_SERVICE_INITIALIZED_EXCEPTION.getCode(),
                        ErrorCode.MODBUS_SERVICE_INITIALIZED_EXCEPTION.getMessage(), e);
            }finally { 
    // workerGroup.shutdownGracefully();
    // bossGroup.shutdownGracefully();
            }
        }

**实质：**  
future.channel().closeFuture().sync();这个语句的主要目的是，**方便测试**，方便写一个非springboot的demo,比如一个简单地junit test方法，closeFuture().sync()可以阻止junit test将server关闭，同时停止test应用的时候也不需要手动再调用关闭服务器的方法workerGroup.shutdownGracefully()…。这样设计在测试时省心。  
但是，当将nettyserver联系到springboot应用的启动时，例如nettyserver设置为@Component,当springboot扫描到nettyserver时，springboot主线程执行到nettyserver的postconstruct注解的方法，然后发生了

future.channel().closeFuture().sync();  
这样导致springboot主线程阻塞，无法继续加载剩下的bean,  
更糟糕的是，如果springboot还添加了springboot-web的依赖（自带tomcat容器），那么被阻塞后将无法启动tomcat servlet engine和webapplicationcontext.

**建议**：

所以不能简单地在nettyserver中的构造方法/init方法中写future.channel().closeFuture().sync();和workerGroup.shutdownGracefully().

只需在构造方法/init方法中bootstrap.bind(port),这是异步的，不会阻塞springboot主线程。

而将stop方法单独抽取出来。

需要注意的是，即使直接关闭springboot应用，不手动调用上面的stop方法，nettyserver也会将之前绑定的端口解除，为了保险起见，可以将stop方法添加@predestroy注解

## 异步模型 ##

### 1.异步基本介绍 ###

1.  异步的概念和同步相对。当一个异步过程调用发出后，调用者不能立刻得到结果。实际处理这个调用的组件在完成后，通过状态、通知和回调来通知调用者。
2.  Netty 中的 I/O 操作是异步的，包括 Bind、Write、Connect 等操作会简单的返回一个 ChannelFuture。
3.  调用者并不能立刻获得结果，而是通过 Future-Listener 机制，用户可以方便的主动获取或者通过通知机制获得 IO 操作结果
4.  Netty 的异步模型是建立在 future 和 callback 的之上的。callback 就是回调。重点说 Future，它的核心思想是：假设一个方法 fun，计算过程可能非常耗时，等待 fun返回显然不合适。那么可以在调用 fun 的时候，立马返回一个 Future，后续可以通过 Future去监控方法 fun 的处理过程(即 ： Future-Listener 机制)

### 2.Future 说明 ###

1.  表示异步的执行结果, 可以通过它提供的方法来检测执行是否完成，比如检索计算等等.
2.  ChannelFuture 是一个接口 ： public interface ChannelFuture extends Future我们可以添加监听器，当监听的事件发生时，就会通知到监听器,代码如下:

ChannelFuture cf = bootstrap.bind(777).sync();
                //给cf 注册监听器，监控我们关心的事件
                cf.addListener(new ChannelFutureListener() { 
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception { 
                        if (cf.isSuccess()) { 
                            System.out.println("监听端口 777 成功");
                        } else { 
                            System.out.println("监听端口 777 失败");
                        }
                    }
                });

### 3.Future-Listener 机制 ###

1.  当 Future 对象刚刚创建时，处于非完成状态，调用者可以通过返回的 ChannelFuture 来获取操作执行的状态，注册监听函数来执行完成后的操作。
2.  常见方法:  
    netty 中的常用的是ChannelFuture 接口，而ChannelFuture 继承了netty的 Future 接口，Future 接口继承了 JDK 中的 Future 接口，然后添加了一些方法：

isDone 方法来判断当前操作是否完成；

**isSuccess** 方法来判断已完成的当前操作是否成功；

getCause 方法来获取已完成的当前操作失败的原因；

isCancelled 方法来判断已完成的当前操作是否被取消；

addListener 方法来注册监听器，当操作已完成(isDone 方法返回完成)，将会通知指定的监听器；如果 Future 对象已完成，则通知指定的监听器

源码

public interface ChannelFuture extends Future<Void> { 
    
        // ChannelFuture 关联的 Channel
        Channel channel();
    
        // 覆写以下几个方法，使得它们返回值为 ChannelFuture 类型 
        @Override
        ChannelFuture addListener(GenericFutureListener<? extends Future<? super Void>> listener);
        @Override
        ChannelFuture addListeners(GenericFutureListener<? extends Future<? super Void>>... listeners);
        @Override
        ChannelFuture removeListener(GenericFutureListener<? extends Future<? super Void>> listener);
        @Override
        ChannelFuture removeListeners(GenericFutureListener<? extends Future<? super Void>>... listeners);
    
        @Override
        ChannelFuture sync() throws InterruptedException;
        @Override
        ChannelFuture syncUninterruptibly();
    
        @Override
        ChannelFuture await() throws InterruptedException;
        @Override
        ChannelFuture awaitUninterruptibly();
    
        // 用来标记该 future 是 void 的，
        // 这样就不允许使用 addListener(...), sync(), await() 以及它们的几个重载方法
        boolean isVoid();
    }

看完上面的 Netty 的 Future 接口，我们可以发现，它加了 sync() 和 await() 用于阻塞等待，还加了 Listeners，只要任务结束去回调 Listener 们就可以了，那么我们就不一定要主动调用 isDone() 来获取状态，或通过 get() 阻塞方法来获取值。  
**sync() 和 await() 的区别**：sync() 内部会先调用 await() 方法，等 await() 方法返回后，会检查下这个任务是否失败，如果失败，重新将导致失败的异常抛出来。也就是说，如果使用 await()，任务抛出异常后，await() 方法会返回，但是不会抛出异常，而 sync() 方法返回的同时会抛出异常。

链接：https://juejin.im/post/5bdfde8251882516f6632dfe

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center]  
优秀链接：  
https://www.jianshu.com/p/23393851dc1a  
https://juejin.im/post/5bdfde8251882516f6632dfe  
https://juejin.im/entry/584fefc6ac502e00693b5bcf

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20200707105408152.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODQ0Njg5MQ==,size_16,color_FFFFFF,t_70#pic_center