netty基础与原理-Toy模板网

这篇具有很好参考价值的文章主要介绍了netty基础与原理。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

Netty线程模型和Reactor模式

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

事件驱动的应用？

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

优点

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

缺点

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

三种模型：

Reactor单线程模型（比较少用）
服务端读请求数据并响应；客户端写请求并读取响应

使用场景:

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

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

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）

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

看起来是单例模式

netty基础与原理,java,网络,netty

辞典

接口和实现类

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 失败以后，返回空结果给消费端，然后由后台线程执行定时任务重试，多用于消息通知这种场景。文章来源地址https://www.toymoban.com/news/detail-645017.html

那么其他的降级是怎么实现的？

feture和Promise?

到了这里，关于netty基础与原理的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

netty基础与原理

Netty线程模型和Reactor模式

Netty快速上手案例

核心链路源码

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

Netty的解码器Decoder和使用场景

Netty编码器Encoder

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

网络传输TCP粘包拆包

TCP半包读写常见解决方案

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

HashedWheelTimer

辞典

接口和实现类

使用案例

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

支付宝扫一扫领取红包，优惠每天领

二维码1

二维码2