Netty线程模型和Reactor模式

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

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

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

优点

1. 响应快，不会因为单个同步而阻塞，虽然Reactor本身依然 是同步的；

2. 编程相对简单，最大程度的避免复杂的多线程及同步问题， 并且避免了多线程/进程的切换开销；

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

缺点

1. 相比传统的简单模型，Reactor增加了一定的复杂性，因而 有一定的门槛，并且不易于调试。
2. 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项目搭建

1. 什么是Echo服务：就是一个应答服务（回显服务器），客户端发 送什么数据，服务端就响应的对应的数据是一个非常有的用于调试 和检测的服务
2. 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() {                // 自定义个handlerprotected 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*/@Overridepublic 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*/@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {System.out.println("EchoServerHandlechannelReadComplete 从管道中读取数据完毕...");}/** 异常捕获* @param: ctx* @param: cause* @return: void*/@Overridepublic 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*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println("Active 管道激活...");ctx.writeAndFlush(Unpooled.copiedBuffer("711 改了密码：",CharsetUtil.UTF_8));}/** 管道数据读取完毕* @param: ctx* @return: void*/@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {System.out.println("EchoClientHandlerchannelReadComplete 管道读取完毕...");}/** 异常捕获* @param: ctx* @param: cause* @return: void*/@Overridepublic 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自带序列化 的缺点 )

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

业界里面也有其他编码框架： 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?