Netty线程模型和Reactor模式

简介：reactor模式和 Netty线程模型

设计模式——Reactor模式（反应器设计模式），是一种基于事件驱动的设计模式，在事件驱动的应用中，将一个或多个客户的服务请求分离（demultiplex）和调度（dispatch）给应用程序。在事件驱动的应用中，同步地、有序地处理同时接收的多个服务请求一般出现在高并发系统中，比如Netty，Redis等。

通俗理解：KTV例子前台接待，服务人员带领去开机器. Reactor模式基于事件驱动，适合处理海量的I/O事件，属于同步非阻塞IO(NIO)

优点

响应快，不会因为单个同步而阻塞，虽然Reactor本身依然是同步的；
编程相对简单，最大程度的避免复杂的多线程及同步问题，并且避免了多线程/进程的切换开销；
可扩展性，可以方便的通过增加Reactor实例个数来充分利用CPU资源；

缺点

相比传统的简单模型，Reactor增加了一定的复杂性，因而有一定的门槛，并且不易于调试。
Reactor模式需要系统底层的的支持，比如Java中的 Selector支持，操作系统的select系统调用支持

Reactor单线程模型（比较少用）

1）作为NIO服务端，接收客户端的TCP连接；作为NIO客户端，向服务端发起TCP连接；

2）服务端读请求数据并响应；客户端写请求并读取响应

使用场景:

对应小业务则适合，编码简单；对于高负载、大并发的应用场景不适合，一个NIO线程处理太多请求，则负载过高，并且可能响应变慢，导致大量请求超时，而且万一线程挂了，则不可用了

Reactor多线程模型

内容：

1）一个Acceptor线程，一组NIO线程，一般是使用自带的线程池，包含一个任务队列和多个可用的线程

使用场景：

可满足大多数场景，但是当Acceptor需要做复杂操作的时候，比如认证等耗时操作，再高并发情况下则也会有性能问题

Reactor主从线程模型

内容:

Acceptor不再是一个线程，而是一组NIO线程；IO线程也是一组NIO线程，这样就是两个线程池去处理接入连接和处理IO
使用场景：满足目前的大部分场景，也是Netty推荐使用的线程模型 BossGroup WorkGroup

补充：

为什么Netty使用NIO而不是AIO，是同步非阻塞还是异步非阻塞？

答：在Linux系统上，AIO的底层实现仍使用EPOLL，与NIO相同，因此在性能上没有明显的优势.

Netty整体架构是reactor模型，采用epoll机制，所以往深的说，还是IO多路复用模式，所以也可说 netty是同步非阻塞模型

很多人说这是netty是基于Java NIO 类库实现的异步通讯框架

特点：异步非阻塞、基于事件驱动，性能高，高可靠性和高可定制性。

Netty快速上手案例

Echo服务和Netty项目搭建

什么是Echo服务：就是一个应答服务（回显服务器），客户端发送什么数据，服务端就响应的对应的数据是一个非常有的用于调试和检测的服务
maven依赖地址：https://mvnrepository.com/artifact/io.netty/netty-all/4.1.32.Final

io.netty netty-all 4.1.32.Final

Echo服务-服务端程序编写对应的启动类和handler处理器:

EchoServer.java

/** 
 * @Description: Echo服务端 
 */ 
public class EchoServer {
    private int port;// 服务端端口号  
    public EchoServer(int port) { 
        this.port = port;
    }
    
    /**   * 启动服务的run 方法   */ 
    public void run() throws InterruptedException { 
        // 配置服务端线程组   
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();        // 老板：后台线程组    
        NioEventLoopGroup workGroup = new NioEventLoopGroup();        // 员工：前台线程组  
        try {   
            ServerBootstrap serverBootstrap = new ServerBootstrap();// 服务启动引导类
            serverBootstrap.group(bossGroup,workGroup)         // 线程组交给服务引导类
                .channel(NioServerSocketChannel.class)         //指定服务端和客户端链接的管道
                .childHandler(         
                new ChannelInitializer() {                // 自定义个handler
                    protected void initChannel(SocketChannel socketChannel) throws Exception {                 							// 添加自己创建的 EchoServerHandler 逻辑                     
                        socketChannel.pipeline().addLast(
                            new EchoServerHandler());    
                    }          
                }); 
            System.out.println("Echo 服务器启动 ing...");       //绑定端口，同步等待成功     
            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();       //等待服务端监听端口关闭
            channelFuture.channel().closeFuture().sync();
       }finally {
            //优雅退出，释放线程池
            workGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
       }
   }
    public static void main(String [] args)
        throws InterruptedException {
        int port = 8080;
        if(args.length > 0){
            port = Integer.parseInt(args[0]);
       }
        new EchoServer(port).run();
   }
}

EchoServerHandler.java

/**
* @Auther: csp1999
* @Date: 2020/09/17/21:15
* @Description: Echo服务端逻辑处理器
*/
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    /*
     * 从管道中读取数据
     * @param: ctx
     * @param: msg
     * @return: void
     */
      @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
        throws Exception {
        ByteBuf data = (ByteBuf) msg;
        System.out.println("服务端收到数据: " + data.toString(CharsetUtil.UTF_8));
        ctx.writeAndFlush(data);
   }
    /*
     * 从管道中读取数据完毕
     * @param: ctx
     * @return: void
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("EchoServerHandlechannelReadComplete 从管道中读取数据完毕...");
   }
    /*
     * 异常捕获
     * @param: ctx
     * @param: cause
     * @return: void
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause) throws Exception {
        cause.printStackTrace();// 打印异常
        ctx.close();
   }
}

EchoClient.java

/** 
 * @Description: echo 客户端 
 */ 
public class EchoClient {  
    private String host;// 客户端ip   
    private int port;// 客户端端口号
    public EchoClient(String host, int port) {
        this.host = host; 
        this.port = port;
    }   
    
    /**   * 启动服务的run 方法   */  
    public void run() throws InterruptedException {  
        // 定义线程组    
        EventLoopGroup group = new NioEventLoopGroup();
        try { 
            Bootstrap bootstrap = new Bootstrap();       // 客户端服务启动引导类 
            bootstrap.group(group)// 线程组交给服务 引导类    
                .channel(NioSocketChannel.class)         // 指定服务端和客户端链接的管道 NioServerSocketChannel
                .remoteAddress(new InetSocketAddress(host, port))// 地址和端口     
                .handler( 
                	new ChannelInitializer() {   
                        protected void initChannel(SocketChannel ch) throws Exception {    
                            // EchoClientHandler 逻辑  
                            ch.pipeline().addLast(new EchoClientHandler());
                        }
                    });
            //连接到服务端，connect是异步连接，在调用同步 等待sync，等待连接成功 
            ChannelFuture channelFuture = bootstrap.connect().sync();       //阻塞直到客户端通道关闭    
            channelFuture.channel().closeFuture().sync();  
        } finally {  
            //优雅退出，释放NIO线程组 
            group.shutdownGracefully();
        }  
    }
    
    public static void main(String[] args) throws InterruptedException { 
        new EchoClient("127.0.0.1", 8080).run();  
    }
}

EchoClientHandler.java

/**
* @Description: echo 客户端逻辑处理器
*/
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        System.out.println("Client received 客户端收到的数据: " + msg.toString(CharsetUtil.UTF_8));
   }
    /*
     * 管道激活
     * @param: ctx
     * @return: void
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Active 管道激活...");
        ctx.writeAndFlush(Unpooled.copiedBuffer("711 改了密码：",CharsetUtil.UTF_8));
   }
    /*
     * 管道数据读取完毕
     * @param: ctx
     * @return: void
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("EchoClientHandlerchannelReadComplete 管道读取完毕...");
   }
    /*
     * 异常捕获
     * @param: ctx
     * @param: cause
     * @return: void
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwablecause) throws Exception {
      cause.printStackTrace();// 抛出异常
      ctx.close();
 }
}

核心链路源码

EventLoop和EventLoopGroup线程模型
1）高性能RPC框架的3个要素：IO模型、数据协议、线程模型。
2）EventLoop 就是一个线程，1个 EventLoop 可以服务多个Channel 管道，1个 Channel 只有一个EventLoop；可以创建多个 EventLoop 来优化资源利用，也就是EventLoopGroup。
3）EventLoopGroup 负责分配 EventLoop 到新创建的Channel。

4）源码分析默认线程池数量 = CPU 核数*2

# EventLoop -> 维护一个 Selector(单线程的方 式管理多个Channel)

学习资料：http://ifeve.com/selectors/

\2. Netty启动引导类Bootstrap模块

简介：Netty启动引导类Bootstrap作用和tcp通道参数设置

参考：https://blog.csdn.net/QH_JAVA/article/details/78383543

ServerBootstrap serverBootstrap = new ServerBootstrap();

group : 设置线程组模型，Reactor 线程模型对比 EventLoopGroup 1. 单线程 2. 多线程 3. 主从线程参考：https://blog.csdn.net/QH_JAVA/article/details/784436 46
channal 设置channel通道类型NioServerSocketChannel、 OioServerSocketChannel
option: 作用于每个新建立的 channel，设置TCP连接中的一些参数，如下：
- ChannelOption.SO_BACKLOG: 存放已完成三次握手的请求的等待队列的最大长度;
- ChannelOption.TCP_NODELAY: 为了解决Nagle的算法问题，默认是false, 要求高实时性，有数据时马上发送，就将该选项设置为true关闭Nagle算法；如果要减少发送次数，就设置为false，会累积一定大小后再发送；
childOption: 作用于被accept之后的连接
childHandler: 用于对每个通道里面的数据处理

Channel模块

简介: Channel 的作用，核心模块，生命周期等

Channel: 客户端和服务端建立的一个连接通道
ChannelHandler：负责Channel的逻辑处理
ChannelPipeline: 负责管理ChannelHandler的有序容器

他们是什么关系：

一个Channel包含一个ChannelPipeline，所有ChannelHandler 都会顺序加入到ChannelPipeline中.

创建Channel时会自动创建一个ChannelPipeline，每个Channel都有一个管理它的pipeline，这关联是永久性的。

Channel当状态出现变化，就会触发对应的事件：

状态：

channelRegistered: channel注册到一个EventLoop
channelActive: 变为活跃状态（连接到了远程主机），可以接受和发送数据
channelInactive: channel处于非活跃状态，没有连接到远程主机
channelUnregistered: channel已经创建，但是未注册到一个 EventLoop里面，也就是没有和Selector绑定

例如:

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    System.out.println("Active 管道激活...");
    ctx.writeAndFlush(Unpooled.copiedBuffer("711改了密码：", CharsetUtil.UTF_8));   
}

ChannelHandler和ChannelPipeline模块

简介：ChannelHandler和ChannelPipeline核心作用和生命周期

方法：

handlerAdded : 当 ChannelHandler 添加到 ChannelPipeline 调用
handlerRemoved : 当 ChannelHandler 从 ChannelPipeline 移除时调用
exceptionCaught : 执行抛出异常时调用

Netty异步操作模块ChannelFuture

Netty中的所有I/O操作都是异步的,这意味着任何I/O调用都会立即返回，而ChannelFuture会提供有关的信息I/O操作的结果或状态。

ChannelFuture状态：

未完成：当I/O操作开始时，将创建一个新的对象，新的最初是未完成的 - 它既没有成功，也没有被取消，因为I/O操作尚未完成。
已完成：当I/O操作完成，不管是成功、失败还是取消， Future都是标记为已完成的, 失败的时候也有具体的信息，例如原因失败，但请注意，即使失败和取消属于完成状态。

注意：不要在IO线程内调用future对象的sync或者await方法，不能在channelHandler中调用sync或者await方法。

ChannelPromise：继承于ChannelFuture，进一步拓展用于设置IO操作的结果

Netty网络数据传输编码与解码

什么是编码、解码

高性能RPC框架的3个要素：IO模型、数据协议、线程模型

最开始接触的编码：java序列化/反序列化（就是编解码）、url编码、base64编解码

为啥jdk有编解码，还要netty自己开发编解码？( java自带序列化的缺点 )

无法跨语言
序列化后的码流太大，也就是数据包太大
序列化和反序列化性能比较差

业界里面也有其他编码框架： google的 protobuf（PB)、 Facebook的Trift、Jboss的Marshalling、Kyro等

Netty里面的编解码：

解码器：负责处理入站 InboundHandler数据
编码器：负责出站 OutboundHandler 数据

Netty里面提供默认的编解码器，也支持自定义编解码器：

Encoder:编码器
Decoder:解码器
Codec:编解码器

Netty的解码器Decoder和使用场景

:Decoder对应的就是ChannelInboundHandler，主要就是字节数组转换为消息对象。

主要是两个方法：

decode ：一般都使用该方式
decodeLast：用于最后的几个字节处理，也就是channel 关闭的时候，产生的最后一个消息。

抽象解码器：

ByteToMessageDecoder：用于将字节转为消息，需要检查缓冲区是否有足够的字节。
ReplayingDecoder：继承ByteToMessageDecoder，不需要检查缓冲区是否有足够的字节，但是R速度略慢于ByteToMessageDecoder，不是所有的ByteBuf都支持。
MessageToMessageDecoder：用于从一种消息解码为另外一种消息（例如：POJO到POJO）

选择：项目复杂性高则使用ReplayingDecoder，否则使用 ByteToMessageDecoder

解码器具体的实现，用的比较多的是(更多是为了解决TCP底层的粘包和拆包问题)： DelimiterBasedFrameDecoder：指定消息分隔符的解码器 xxx&aaa&bbb

LineBasedFrameDecoder: 以换行符为结束标志的解码器
FixedLengthFrameDecoder：固定长度解码器
LengthFieldBasedFrameDecoder：message = header+body, 基于长度解码的通用解码器
StringDecoder：文本解码器，将接收到的对象转化为字符串，一般会与上面的进行配合，然后在后面添加业务handle

Netty编码器Encoder

Encoder对应的就是ChannelOutboundHandler，消息对象转换为字节数组。

Netty本身未提供和解码一样的编码器，是因为场景不同，两者非对等的：

MessageToByteEncoder：消息转为字节数组，调用write方法，会先判断当前编码器是否支持需要发送的消息类型，如果不支持，则透传；
MessageToMessageEncoder：用于从一种消息编码为另外一种消息（例如POJO到POJO）

数据协议处理之Netty编解码器类Codec

组合解码器和编码器，以此提供对于字节和消息都相同的操作。

优点：成对出现，编解码都是在一个类里面完成。

缺点：耦合在一起，拓展性不佳

Codec:组合编解码
- ByteToMessageCodec
- MessageToMessageCodec
decoder:解码
- ByteToMessageDecoder
- MessageToMessageDecoder
encoder:编码
- ByteToMessageEncoder
- MessageToMessageEncoder

网络传输TCP粘包拆包

什么是TCP粘包拆包?

TCP拆包: 一个完整的包可能会被TCP拆分为多个包进行发送

TCP粘包: 把多个小的包封装成一个大的数据包发送, client发送的若干数据包 Server接收时粘成一包

发送方和接收方都可能出现这种情况：

发送方的原因：TCP默认会使用Nagle算法

接收方的原因: TCP接收到数据放置缓存中，应用程序从缓存中读取

UDP: 是没有粘包和拆包的问题，有边界协议

TCP半包读写常见解决方案

发送方：可以关闭Nagle算法

接受方：TCP是无界的数据流，并没有处理粘包现象的机制, 且协议本身无法避免粘包，半包读写的发生需要在应用层进行处理，应用层解决半包读写的办法。

设置定长消息 (10字符)： xdclass000xdclass000xdclass000xdclass000
设置消息的边界 ($$ 切割)： sdfafwefqwefwe$$dsafadfadsfwqehidwuehfiw$$879329832 r89qweew$$
使用带消息头的协议，消息头存储消息开始标识及消息的长度信息：Header+Body

Netty自带解决TCP半包读写方案

DelimiterBasedFrameDecoder：指定消息分隔符的解码器
LineBasedFrameDecoder: 以换行符为结束标志的解码器
FixedLengthFrameDecoder：固定长度解码器(Netty 使用该解码器)
LengthFieldBasedFrameDecoder：message = header+body, 基于长度解码的通用解码器
herzbeat的笑脸?

HashedWheelTimer

时间轮算法管理延迟任务队列

在网络通信中管理上万的连接，每个连接都有超时任务，如果为每个任务启动一个TImer超时器，那么会占用大量资源。为了解决这个问题，可用Netty工具类HashedWheelTimer。

类似前端setTimeout那样的超时任务？
为什么每个连接都超时任务的话占用大量资源?一个单独的又好在哪里？网络通信中有什么地方需要超时任务？can can netty
超超我的
为什么网络连接有超时任务？

这个类用来计划执行非精准的I/O超时。可以通过指定每一格的时间间隔来改变执行时间的精确度。在大多数网络应用中，I/O超时不需要十分准确，因此，默认的时间间隔是100 毫秒，这个值适用于大多数场合。HashedWheelTimer内部结构可以看做是个车轮，简单来说，就是TimerTask的hashTable的车轮。车轮的size默认是512，可以通过构造函数自己设置这个值。注意，当HashedWheelTimer被实例化启动后，会创建一个新的线程，因此，你的项目里应该只创建它的唯一一个实例。

（这个类源自一位教授的论文 Tony Lauck）

意思是说，时间间隔越小精度越高？因为没必要完全精准，所以用时间轮每隔一段时间去处理？

看起来是单例模式

辞典

接口和实现类

Timer 计时器，负责总管TimeOut和TimeTask
- HashedWheelTimer 将任务散列成一个环（底层是数组）来管理，每个散列桶双向链表串联TimeTask
TimeOut 一段延迟后只执行一次的任务，类似前端的setTimeout任务
TimeTask 被设置于TimeOut中的具体的任务

tick: 时钟“嘀嗒”一声，代表时间轮转动一下，到下一个hash桶

提交任务的线程，只要把任务往虚线上面的任务队列中存放即可返回。工作线程是单线程，一旦开启，不停地在时钟上绕圈圈。

超时任务设置为1000毫秒，超时之后，由hashedWheelTimer类中的worker线程，执行超时之后的任务。

hashedWheelTimer有32个槽（类比HashMap中的桶），默认每100毫秒移动下一个槽。
任务需要经过的tick数为: 1000 / 100 = 10次 (等待时长 / tickDuration)
任务需要经过的轮数为 : 10次 / 32次/轮 = 0轮 (tick总次数 / ticksPerWheel)

因为任务超时后不能马上被worker线程执行，需要等到worker线程移到相应卡槽位置时才会执行，因此说执行时间不精确。

啊？那我们刚才算的是个什么啊？

hashedWheelTimer的核心是Worker线程，主要负责每过tickDuration时间就累加一次tick. 同时, 也负责执行到期的timeout任务, 此外，还负责添加timeou任务到指定的wheel中。

TimerTask 非常简单，就一个 run() 方法：

public interface TimerTask {
    void run(Timeout timeout) throws Exception;
}

这里有点意思的是，它把 Timeout 的实例也传进来了，我们平时的代码习惯，都是单向依赖。

这样做也有好处，那就是在任务执行过程中，可以通过 timeout 实例来做点其他的事情。

Timeout 也是一个接口类：

public interface Timeout {
    Timer timer();
    TimerTask task();
    boolean isExpired();
    boolean isCancelled();
    boolean cancel();
}

它持有上层的 Timer 实例，和下层的 TimerTask 实例，然后取消任务的操作也在这里面。

使用案例

Dubbo 的集群调用策略 FailbackClusterInvoker 中：

它在调用 provider 失败以后，返回空结果给消费端，然后由后台线程执行定时任务重试，多用于消息通知这种场景。

那么其他的降级是怎么实现的？
feture和Promise?

netty基础与原理

Netty线程模型和Reactor模式

Netty快速上手案例

核心链路源码

Netty网络数据传输编码与解码

Netty的解码器Decoder和使用场景

Netty编码器Encoder

数据协议处理之Netty编解码器类Codec

网络传输TCP粘包拆包

TCP半包读写常见解决方案

Netty自带解决TCP半包读写方案

HashedWheelTimer

辞典

接口和实现类

使用案例

相关文章

跑步运动耳机哪个牌子好、跑步运动耳机推荐

开工大吉|华润鞋业二期自动化改造项目开工典礼圆满举行

从 Zebec Protocol 长期布局看，ZBC 通证的潜在应用场景

题目大解析（3）

《算法竞赛·快冲300题》每日一题：“造电梯”

使用python对图像加噪声

Arduino ESP32 v2 使用记录：开发环境搭建

写一个函数返回参数二进制中 1 的个数(c语言三种实现方法)

Kubernetes 调度约束(亲和性、污点、容忍）

SAP MM学习笔记18- SQVI 工具

【hello C++】智能指针

10.pod资源限制和健康检查

5. vue-element-admin 二次开发攻略指南

计算机网络-专业术语

STM32CubeMX之freeRTOS信号量

Hadoop+Python+Django+Mysql热门旅游景点数据分析系统的设计与实现（包含设计报告）

人工智能讲师AIGC讲师叶梓:大模型这么火，我们在使用时应该关注些什么？-2

java程序打包成exe在无java环境执行

C#随机法双峰函数求极值避免落入局部最优解

【学会动态规划】买卖股票的最佳时机 III（17）