【Netty】线程模型

追求适度,才能走向成功;人在顶峰,迈步就是下坡;身在低谷,抬足既是登高;弦,绷得太紧会断;人,思虑过度会疯;水至清无鱼,人至真无友,山至高无树;适度,不是中庸,而是一种明智的生活态度。

导读:本篇文章讲解 【Netty】线程模型,希望对大家有帮助,欢迎收藏,转发!站点地址:www.bmabk.com,来源:原文


单线程模型 (单Reactor单线程)
多线程模型 (单Reactor多线程)
主从多线程模型 (多Reactor多线程)

1. 单Reactor单线程

在这里插入图片描述
所有操作都在同一个NIO线程处理,在这个单线程中要负责接收请求,处理IO,编解码所有操作,相当于一个饭馆只有一个人,同时负责前台和后台服务,效率低。

  1. 一个NIO线程同时处理成百上千的连接,性能上无法支撑,即便NIO线程的CPU负荷达到100%,也无法满足海量消息的编码、解码、读取和发送。
    2.当NIO线程负载过重之后,处理速度将变慢,这会导致大量客户端连接超时,超时之后往往会进行重发,这更加重了NIO线程的负载,最终会导致大量消息积压和处理超时,成为系统的性能瓶颈。
    3.可靠性问题:一旦NIO线程意外跑飞,或者进入死循环,会导致整个系统通信模块不可用,不能接收和处理外部消息,造成节点故障。

使用场景:客户端发送连接数量有限,且业务线程处理时间非常快 可以使用单个线程维护多个不同连接 ,代表有Redis。

Netty相关代码

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
ServerBootstrap bootstrap = new ServerBootstrap();

优点:
不需要上下文切换 、不存在线程安全的问题

缺点:
1.只有一个线程处理,无法发挥cpu多核的效率
2.如果请求比较多的情况下,容易遇到瓶颈
3.某种意外的原因线程终止了,导致整个系统模块无法使用。

适用于客户端连接数量有一定限制,业务处理时间非常快。

2. 单Reactor多线程

在这里插入图片描述
相当于一个饭馆有一个前台负责接待,有很多服务员去做后面的工作,这样效率就比单线程模型提高很多。

1.由一个Reactor线程-Acceptor线程用于监听服务端,接收客户端连接请求;
2.网络I/O操作读、写等由Reactor线程池负责处理;
3.一个Reactor线程可同时处理多条链路,但一条链路只能对应一个Reactor线程,这样可避免并发操作问题;
4.绝大多数场景下,Reactor多线程模型都可以满足性能需求,但是,在极个别特殊场景中,一个Reactor线程负责监听和处理所有的客户端连接可能会存在性能问题。例如并发百万客户端连接,或者服务端需要对客户端握手进行安全认证,但是认证本身非常损耗性能。因此,诞生了第三种线程模型;

Netty设置相关代码

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup(2);
ServerBootstrap bootstrap = new ServerBootstrap();

创建1个线程的bossGroup线程组,这个线程负责处理客户端的连接请求,而workerGroup默认使用处理器个数*2的线程数量来处理I/O操作。这就相当于Reactor的多线程模型。

优点
可以充分利用多线程多核处理能力 ,处理效率比单线程要快

缺点:
对线程之间访问数据,有可能遇到线程安全问题。

3. Reactor主从模型

多线程模型的缺点在于并发量很高的情况下,只有一个Reactor单线程去处理是来不及的,就像饭馆只有一个前台接待很多客人也是不够的。为此需要使用主从线程模型。
主从线程模型:一组线程池接收请求,一组线程池处理IO。

1.服务端使用一个独立的主Reactor线程池来处理客户端连接,当服务端收到连接请求时,从主线程池中随机选择一个Reactor线程作为Acceptor线程处理连接;
2.链路建立成功后,将新创建的SocketChannel注册到sub reactor线程池的某个Reactor线程上,由它处理后续的I/O操作。

1.mainReactor负责监听server socket,用来处理新连接的建立,将建立的socketChannel指定注册给subReactor。
2.subReactor维护自己的selector, 基于mainReactor 注册的socketChannel多路分离IO读写事件,读写网 络数据,对业务处理的功能,另其扔给worker线程池来完成。

通俗易懂理解为:mainReactor为老板、只负责接口、subReactor负责前台接待,也就是多个前台接待。

比起多线程单 Rector 模型,它是将 Reactor 分成两部分,mainReactor 负责监听并 Accept新连接,然后将建立的 socket 通过多路复用器(Acceptor)分派给subReactor。subReactor 负责多路分离已连接的 socket,读写网络数据;业务处理功能,其交给 worker 线程池完成。通常,subReactor 个数上可与 CPU 个数等同。

4. Netty线程模型

在这里插入图片描述
1.创建服务端的时候实例化了 2 个 EventLoopGroup。bossGroup 线程组实际就是 Acceptor 线程池,负责处理客户端的 TCP 连接请求。workerGroup 是真正负责 I/O 读写操作的线程组。通过这里能够知道 Netty 是多 Reactor 模型。

2.ServerBootstrap 类是 Netty 用于启动 NIO 的辅助类,能够方便开发。通过 group 方法将线程组传递到 ServerBootstrap 中,设置 Channel 为 NioServerSocketChannel,接着设置 NioServerSocketChannel 的 TCP 参数,最后绑定 I/O 事件处理类 ChildChannelHandler。
辅助类完成配置之后调用 bind 方法绑定监听端口,Netty 返回 ChannelFuture,f.channel().closeFuture().sync() 对同步阻塞的获取结果。

3.调用线程组 shutdownGracefully 优雅推出,释放资源。

1.Netty核心抽象出:两个线程池bossGroup和workerGroup;
2.bossGroup负责事件的接收,workerGroup负责处理该时间的IO流读写;
3.NioEventLoopGroup类似于是一个事件循环组,该组有有多个不同的事件循环 NioEventLoop 表示一个不断循环的执行处理任务的线程, 每个 NioEventLoop 都有一个 Selector , 用于监听绑定在其上的 Socket 的网络通讯;
4.new NioEventLoopGroup(BossGroup(接收线程)、WorkerGroup(工作线程)) 可以有多个线程, 即可以含有多个 NioEventLoop;
5.BossGroup(接收线程) 负责轮训接收accept事件,并且其注册到 Worker 的 NIOEventLoop 上的 Selector,处理该队列任务;
6.WorkerGroup(工作线程) 轮询read, write 事件 处理IO事件 在找到对应的NioScocketChannel 处理;
默认的情况下 BossGroup 和 WorkerGroup 子线程数是:当前cpu核数*2
在这里插入图片描述
1.ChannelHandlerContex 保存 Channel 相关的所有上下文信息,同时关联一个 ChannelHandler 对象;
2. Pipeline 管道 设置 编码与解码 handler 采用双向链表存放
在这里插入图片描述

4.1 TaskQueue用法

为了防止 某个Handler 中有个长时间的操作,会造成 Pipeline管道的阻塞 可以使用taskQueue提交任务异步执行。

            try {

                ByteBuf buf = (ByteBuf) msg;
                byte[] req = new byte[buf.readableBytes()];
                buf.readBytes(req);
                String body = new String(req, "UTF-8");
                System.out.println(Thread.currentThread().getName() + "服务器端接收到请求,读取到数据 : " + body);

                ctx.channel().eventLoop().execute(() -> {
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //异步发送应答消息给客户端: 这里并没有把消息直接写入SocketChannel,而是放入发送缓冲数组中
                    ByteBuf resp = Unpooled.copiedBuffer("www.baidu.com".getBytes());
                    System.out.println(Thread.currentThread().getName() + "服务器端响应数据给客户端");
                    ctx.writeAndFlush(resp);
                });

            } catch (Exception e) {
                e.printStackTrace();
            }

ctx.channel().eventLoop().schedule(() -> {} 设置可定时执行

4.2 心跳机制

心跳机制 判断服务器端与客户端存活

import io.netty.bootstrap.ServerBootstrap;
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.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;

import java.net.SocketAddress;


/**
 * 服务端代码
 */
public class HeartBeatServer {


    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap()
                .group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new StringEncoder());
                        pipeline.addLast(new StringDecoder());
                        //IdleStateHandler的readerIdleTime参数指定超过3秒还没收到客户端的连接,
                        //会触发IdleStateEvent事件并且交给下一个handler处理,下一个handler必须
                        //实现userEventTriggered方法处理对应事件
                        /**
                         * readerIdleTimeSeconds  读操作空闲3s
                         * writerIdleTimeSeconds  写操作空闲0秒
                         * allIdleTimeSeconds  读写全部 0
                         */
                        pipeline.addLast(new IdleStateHandler(3, 0, 0));
                        pipeline.addLast(new HeartBeatServerHandler());
                    }
                });
        ChannelFuture future = bootstrap.bind(9000).sync();
        System.out.println("netty server start successfully");
        future.channel().closeFuture().sync();
    }
}

class HeartBeatServerHandler extends SimpleChannelInboundHandler<String> {

    private int readIdleTimes = 0;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        System.out.println("====== > [server] message received: " + msg);
        ctx.channel().writeAndFlush("ok");
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        IdleStateEvent event = (IdleStateEvent) evt;
        String eventType;
        switch (event.state()) {
            case READER_IDLE:
                eventType = "读空闲";
                //读空闲的计数加1
                readIdleTimes++;
                break;
            case WRITER_IDLE:
                eventType = "写空闲";
                // 不处理
                break;
            case ALL_IDLE:
                eventType = "读写空闲";
                // 不处理
                break;
            default:
                throw new IllegalStateException("非法状态!");
        }
        System.out.println(ctx.channel().remoteAddress() + "超时事件:" + eventType);
        if (readIdleTimes > 3) {
            System.out.println("[server]读空闲超过3次,关闭连接,释放更多资源");
            ctx.channel().writeAndFlush("idle close");
            ctx.channel().close();
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("=== " + ctx.channel().remoteAddress() + " is active ===");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        SocketAddress address = channel.remoteAddress();
        System.out.println(address + " 下线了");

    }
}


5. Netty核心Api

1.Bootstrap 意思是引导,一个 Netty 应用通常由一个 Bootstrap 开始,主要作用是配置整个 Netty 程序,串联各个组件,Netty 中 Bootstrap 类是客户端程序的启动引导类,ServerBootstrap 是服务端启动引导类
2.常见的方法有
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup),该方法用于服务器端,用来设置两个 EventLoop;
public B group(EventLoopGroup group) ,该方法用于客户端,用来设置一个 EventLoop public B channel(Class<? extends C> channelClass),该方法用来设置一个服务器端的通道实现;
public B option(ChannelOption option, T value),用来给 ServerChannel 添加配置;
public ServerBootstrap childOption(ChannelOption childOption, T value),用来给接收到的通道添加配置;
public ServerBootstrap childHandler(ChannelHandler childHandler),该方法用来设置业务处理类(自定义的 handler);
public ChannelFuture bind(int inetPort) ,该方法用于服务器端,用来设置占用的端口号;
public ChannelFuture connect(String inetHost, int inetPort) ,该方法用于客户端,用来连接服务器。

5.1 Future、ChannelFuture

1.Netty 中所有的 IO 操作都是异步的,不能立刻得知消息是否被正确处理。但是可以过一会等它执行完成或者直接注册一个监听,具体的实现就是通过 Future 和 ChannelFutures,他们可以注册一个监听,当操作执行成功或失败时监听会自动触发注册的监听事件
常见的方法有:
2.Channel channel(),返回当前正在进行 IO 操作的通道
ChannelFuture sync(),等待异步操作执行完毕

5.2 Channel

1.Netty 网络通信的组件,能够用于执行网络 I/O 操作;
2.通过Channel 可获得当前网络连接的通道的状态;
3.通过Channel 可获得 网络连接的配置参数 (例如接收缓冲区大小);
Channel 提供异步的网络 I/O 操作(如建立连接,读写,绑定端口),异步调用意味着任何 I/O 调用都将立即返回,并且不保证在调 用结束时所请求的 I/O 操作已完成;
调用立即返回一个 ChannelFuture 实例,通过注册监听器到 ChannelFuture 上,可以 I/O 操作成功、失败或取消时回调通知调用方;
4.支持关联 I/O 操作与对应的处理程序 不同协议、不同的阻塞类型的连接都有不同的 Channel 类型与之对应;
5.常用的 Channel 类型:
NioSocketChannel,异步的客户端 TCP Socket 连接;
NioServerSocketChannel,异步的服务器端 TCP Socket 连接;
NioDatagramChannel,异步的 UDP 连接;
NioSctpChannel,异步的客户端 Sctp 连接;
NioSctpServerChannel,异步的 Sctp 服务器端连接,这些通道涵盖了 UDP 和 TCP 网络 IO 以及文件 IO。

5.3 Selector

1.Netty 基于 Selector 对象实现 I/O 多路复用,通过 Selector 一个线程可以监听多个连接的 Channel 事件;
2.当向一个 Selector 中注册 Channel 后,Selector 内部的机制就可以自动不断地查询(Select) 这些注册的 Channel 是否有已就绪的 I/O 事件(例如可读,可写,网络连接完成等),这样程序就可以很简单地使用一个线程高效地管理多个 Channel。

5.4 ChannelHandler 及其实现类

ChannelHandler 是一个接口,处理 I/O 事件或拦截 I/O 操作,并将其转发到其 ChannelPipeline(业务处理链)中的下一个处理程序。
ChannelHandler 本身并没有提供很多方法,因为这个接口有许多的方法需要实现,方便使用期间,可以继承它的子类

5.5 Pipeline 和 ChannelPipeline

ChannelPipeline 是一个 Handler 的集合,它负责处理和拦截 inbound 或者 outbound 的事件和操作,相当于一个贯穿 Netty 的链。(也可以这样理解:ChannelPipeline 是 保存 ChannelHandler 的 List,用于处理或拦截 Channel 的入站事件和出站操作)
ChannelPipeline 实现了一种高级形式的拦截过滤器模式,使用户可以完全控制事件的处理方式,以及 Channel 中各个的 ChannelHandler 如何相互交互

5.6 ChannelHandlerContex

1.保存 Channel 相关的所有上下文信息,同时关联一个 ChannelHandler 对象;
2.ChannelHandlerContext 中 包 含 一 个 具 体 的 事 件 处 理 器 ChannelHandler , 同 时ChannelHandlerContext 中也绑定了对应的 pipeline 和 Channel 的信息,方便对 ChannelHandler进行调用.
常用方法 :
1)ChannelFuture close(),关闭通道
2)ChannelOutboundInvoker flush(),刷新
3)ChannelFuture writeAndFlush(Object msg) , 将 数 据 写 到 ChannelPipeline 中 当 前
4)ChannelHandler 的下一个 ChannelHandler 开始处理(出站)

5.7 ChannelOption

1.Netty 在创建 Channel 实例后,一般都需要设置 ChannelOption 参数。
2.ChannelOption 参数如下:
ChannelOption.SO_BACKLOG–对应 TCP/IP 协议 listen 函数中的 backlog 参数,用来初始化服务器可连接队列大小。服 务端处理客户端连接请求是顺序处理的,所以同一时间只能处理一个客户端连接。多个客户端来的时候,服务端将不能处理的客户端连接请求放在队列中等待处理,backlog 参数指定了队列的大小;
ChannelOption.SO_KEEPALIVE–一直保持连接活动状态.

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

文章由极客之音整理,本文链接:https://www.bmabk.com/index.php/post/131193.html

(0)
飞熊的头像飞熊bm

相关推荐

发表回复

登录后才能评论
极客之音——专业性很强的中文编程技术网站,欢迎收藏到浏览器,订阅我们!