3.Netty中Channel通道概述

news2024/11/23 15:06:11

Selector 模型

Java NIO 是基于 Selector 模型来实现非阻塞的 I/O。Netty 底层是基于 Java NIO 实现的,因此也使用了 Selector 模型。

Selector 模型解决了传统的阻塞 I/O 编程一个客户端一个线程的问题。Selector 提供了一种机制,用于监视一个或多个 NIO 通道,并识别何时可以使用一个或多个 NIO 通道进行数据传输。这样,一个线程就可以管理多个通道,从而管理多个网络连接。

image.png
Selector 提供了选择执行已经就绪Channel的任务的能力。从底层来看,Selector 会轮询 Channel 是否已经准备好执行每个 I/O 操作。Selector 允许单线程处理多个 Channel 。Selector 是一种多路复用的技术。

SelectableChannel

并不是所有的 Channel 都是可以被 Selector 复用的,只有抽象类 SelectableChannel的子类才能被 Selector 复用。

例如,FileChannel 就不能被选择器复用,因为 FileChannel 不是SelectableChannel的子类。

为了与 Selector 一起使用,SelectableChannel必须首先通过register方法来注册此类的实例。此方法返回一个新的SelectionKey对象,该对象表示Channel已经在Selector进行了注册。向Selector注册后,Channel将保持注册状态,直到注销为止。

一个 Channel 最多可以使用任何一个特定的 Selector 注册一次,但是相同的 Channel 可以注册到多个 Selector 上。可以通过调用 isRegistered方法来确定是否向一个或多个 Selector 注册了 Channel。

SelectableChannel可以安全的供多个并发线程使用。

Channel 注册到 Selector

使用 SelectableChannelregister方法,可将Channel注册到Selector。方法接口源码如下:

    public final SelectionKey register(Selector sel, int ops)
        throws ClosedChannelException
    {
        return register(sel, ops, null);
    }
    
    public abstract SelectionKey register(Selector sel, int ops, Object att) throws ClosedChannelException;

其中各选项说明如下:

  • sel:指定 Channel 要注册的 Selector
  • ops : 指定 Selector需要查询的通道的操作。

一个Channel在Selector注册其代表的是一个SelectionKey事件,SelectionKey的类型包括:

  • OP_READ:可读事件;值为:1<<0
  • OP_WRITE:可写事件;值为:1<<2
  • OP_CONNECT:客户端连接服务端的事件(tcp连接),一般为创建SocketChannel客户端channel;值为:1<<3
  • OP_ACCEPT:服务端接收客户端连接的事件,一般为创建ServerSocketChannel服务端channel;值为:1<<4

具体的注册代码如下:

 // 1.创建通道管理器(Selector)
 Selector selector = Selector.open();
 
 // 2.创建通道ServerSocketChannel
 ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
 
 // 3.channel要注册到Selector上就必须是非阻塞的,所以FileChannel是不可以使用Selector的,因为FileChannel是阻塞的
 serverSocketChannel.configureBlocking(false);
 
 // 4.第二个参数指定了我们对 Channel 的什么类型的事件感兴趣
 SelectionKey key = serverSocketChannel.register(selector , SelectionKey.OP_READ);
 
 // 也可以使用或运算|来组合多个事件,例如
 SelectionKey key = serverSocketChannel.register(selector , SelectionKey.OP_READ | SelectionKey.OP_WRITE);

值得注意的是:一个 Channel 仅仅可以被注册到一个Selector 一次, 如果将 Channel 注册到 Selector 多次, 那么其实就是相当于更新 SelectionKey的 interest set

SelectionKey

Channel 和 Selector 关系确定后之后,并且一旦 Channel 处于某种就绪状态,就可以被选择器查询到。这个工作再调用 Selector 的 select 方法完成。select 方法的作用,就是对感兴趣的通道操作进行就绪状态的查询。

// 当注册事件到达时,方法返回,否则该方法会一直阻塞
selector.select();

SelectionKey 包含了 interest 集合,代表了所选择的感兴趣的事件集合。可以通过 SelectionKey 读写 interest 集合,例如:

// 返回当前感兴趣的事件列表
int interestSet = key.interestOps();

// 也可通过interestSet判断其中包含的事件
boolean isInterestedInAccept  = interestSet & SelectionKey.OP_ACCEPT;
boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;
boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;
boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;    

// 可以通过interestOps(int ops)方法修改事件列表
key.interestOps(interestSet | SelectionKey.OP_WRITE);

可以看到,用位与操作 interest 集合和给定的 SelectionKey 常量,可以确定某个确定的事件是否在 interest 集合中。

SelectionKey 包含了ready集合。ready 集合是通道已经准备就绪的操作的集合。在一次选择之后,会首先访问这个 ready 集合。可以这样访问 ready 集合:

int readySet = key.readyOps();

// 也可通过四个方法来分别判断不同事件是否就绪
key.isReadable();    //读事件是否就绪
key.isWritable();    //写事件是否就绪
key.isConnectable(); //客户端连接事件是否就绪
key.isAcceptable();  //服务端连接事件是否就绪

我们可以通过SelectionKey来获取当前的channelselector

//返回当前事件关联的通道,可转换的选项包括:`ServerSocketChannel`和`SocketChannel`
Channel channel = key.channel();

//返回当前事件所关联的Selector对象
Selector selector = key.selector();

可以将一个对象或者其他信息附着到 SelectionKey 上,这样就能方便地识别某个特定的通道。

key.attach(theObject);
Object attachedObj = key.attachment();

还可以在用 register()方法向 Selector 注册 Channel 的时候附加对象。

SelectionKey key = channel.register(selector, SelectionKey.OP_READ, theObject);

遍历 SelectionKey

一旦调用了 select 方法,并且返回值表明有一个或更多个通道就绪了,然后可以通过调用 selector的 selectedKey()方法,访问 SelectionKey 集合中的就绪通道,如下所示:

Set<SelectionKey> selectionKeys = selector.selectedKeys();

可以遍历这个已选择的键集合来访问就绪的通道,代码如下:

// 获取监听事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeys.iterator();
// 迭代处理
while (iterator.hasNext()) {
    // 获取事件
    SelectionKey key = iterator.next();
    // 移除事件,避免重复处理
    iterator.remove();
    // 可连接
    if (key.isAcceptable()) {
        ...
    } 
    // 可读
    if (key.isReadable()) {
        ...
    }
    //可写
    if(key.isWritable()){
    	...                
    }
}

事件驱动

Netty是一款异步的事件驱动的网络应用程序框架。在 Netty 中,事件是指对某些操作感兴趣的事。例如,在某个Channel注册了 OP_READ,说明该 Channel 对读感兴趣,当 Channel 中有可读的数据时,它会得到一个事件的通知。

在 Netty 事件驱动模型中包括以下核心组件。

Channel

Channel(管道)是 Java NIO 的一个基本抽象,代表了一个连接到如硬件设备、文件、网络 socket 等实体的开放连接,或者是一个能够完成一种或多种不同的I/O 操作的程序。

回调

回调 就是一个方法,一个指向已经被提供给另外一个方法的方法的引用。这使得后者可以在适当的时候调用前者,Netty 在内部使用了回调来处理事件;当一个回调被触发时,相关的事件可以被一个ChannelHandler接口处理。

例如:在上一篇文章中,Netty 开发的服务端的管道处理器代码中,当Channel中有可读的消息时,NettyServerHandler的回调方法channelRead就会被调用。

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    //读取数据实际(这里我们可以读取客户端发送的消息)
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server ctx =" + ctx);
        Channel channel = ctx.channel();
        //将 msg 转成一个 ByteBuf
        //ByteBuf 是 Netty 提供的,不是 NIO 的 ByteBuffer.
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("客户端发送消息是:" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端地址:" + channel.remoteAddress());
    }


    //处理异常, 一般是需要关闭通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

Future

Future 可以看作是一个异步操作的结果的占位符;它将在未来的某个时刻完成,并提供对其结果的访问,Netty 提供了 ChannelFuture 用于在异步操作的时候使用,每个 Netty 的出站 I/O 操作都将返回一个 ChannelFuture(完全是异步和事件驱动的)。

以下是一个 ChannelFutureListener使用的示例。

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ChannelFuture future = ctx.channel().close();
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                //..
            }
        });
    }

事件及处理器

在 Netty 中事件按照出/入站数据流进行分类:

入站数据或相关状态更改触发的事件包括:

  • 连接已被激活或者失活。
  • 数据读取。
  • 用户事件。
  • 错误事件,

出站事件是未来将会出发的某个动作的操作结果:

  • 打开或者关闭到远程节点的连接。
  • 将数据写或者冲刷到套接字。

每个事件都可以被分发给ChannelHandler类中的某个用户实现的方法。如下图展示了一个事件是如何被一个这样的ChannelHandler链所处理的。

image.png
ChannelHandler 为处理器提供了基本的抽象,可理解为一种为了响应特定事件而被执行的回调。

责任链模式

责任链模式(Chain of Responsibility Pattern)是一种行为型设计模式,它为请求创建了一个处理对象的链。其链中每一个节点都看作是一个对象,每个节点处理的请求均不同,且内部自动维护一个下一节点对象。当一个请求从链式的首端发出时,会沿着链的路径依次传递给每一个节点对象,直至有对象处理这个请求为止。

责任链模式的重点在这个 "链"上,由一条链去处理相似的请求,在链中决定谁来处理这个请求,并返回相应的结果。在Netty中,定义了ChannelPipeline接口用于对责任链的抽象。

责任链模式会定义一个抽象处理器(Handler)角色,该角色对请求进行抽象,并定义一个方法来设定和返回对下一个处理器的引用。在Netty中,定义了ChannelHandler接口承担该角色。

责任链模式的优缺点

优点:

  • 发送者不需要知道自己发送的这个请求到底会被哪个对象处理掉,实现了发送者和接受者的解耦。
  • 简化了发送者对象的设计。
  • 可以动态的添加节点和删除节点。

缺点:

  • 所有的请求都从链的头部开始遍历,对性能有损耗。
  • 不方便调试。由于该模式采用了类似递归的方式,调试的时候逻辑比较复杂。

使用场景:

  • 一个请求需要一系列的处理工作。
  • 业务流的处理,例如文件审批。
  • 对系统进行扩展补充。

ChannelPipeline

Netty 的ChannelPipeline设计,就采用了责任链设计模式, 底层采用双向链表的数据结构,,将链上的各个处理器串联起来。

客户端每一个请求的到来,Netty都认为,ChannelPipeline中的所有的处理器都有机会处理它,因此,对于入栈的请求,全部从头节点开始往后传播,一直传播到尾节点(来到尾节点的msg会被释放掉)。


入站事件:通常指 IO 线程生成了入站数据(通俗理解:从 socket 底层自己往上冒上来的事件都是入站)。
比如EventLoop收到selectorOP_READ事件,入站处理器调用socketChannel.read(ByteBuffer)接受到数据后,这将导致通道的ChannelPipeline中包含的下一个中的channelRead方法被调用。

出站事件:通常指 IO 线程执行实际的输出操作(通俗理解:想主动往 socket 底层操作的事件的都是出站)。
比如bind方法用意时请求server socket绑定到给定的SocketAddress,这将导致通道的ChannelPipeline中包含的下一个出站处理器中的bind方法被调用。

将事件传递给下一个处理器

处理器必须调用ChannelHandlerContext中的事件传播方法,将事件传递给下一个处理器。

入站事件和出站事件的传播方法如下图所示:

image.png
以下示例说明了事件传播通常是如何完成的:

public class MyInboundHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("Connected!");
        ctx.fireChannelActive();
    }
}

public class MyOutboundHandler extends ChannelOutboundHandlerAdapter {

    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        System.out.println("Closing...");
        ctx.close(promise);
    }
}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/1181733.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

如何成为C++大神?五个技巧助你提升编程水平

一名优秀的C程序员是如何炼成的&#xff1f;这个问题一直困扰着许多人&#xff0c;尤其是那些刚刚踏入编程的世界的新手。C作为一门强大而复杂的编程语言&#xff0c;的确需要一些特殊的技巧和策略才能掌握。但幸运的是&#xff0c;成为一名出色的C程序员并不是不可能的任务。在…

【算法练习Day41】买卖股票的最佳时机买卖股票的最佳时机 II

​&#x1f4dd;个人主页&#xff1a;Sherry的成长之路 &#x1f3e0;学习社区&#xff1a;Sherry的成长之路&#xff08;个人社区&#xff09; &#x1f4d6;专栏链接&#xff1a;练题 &#x1f3af;长路漫漫浩浩&#xff0c;万事皆有期待 文章目录 买卖股票的最佳时机买卖股票…

办公神器!2024年值得拥有的10款在线画板软件!

随着科技的进步和互联网的普及&#xff0c;我们工作、学习和生活方式发生了翻天覆地的变化。在线画板软件就是在这个背景下应运而生的一种便捷工具。它不仅满足了我们随时随地绘制图像、演示思路的需求&#xff0c;还提供了协同编辑、云存储等功能&#xff0c;使得团队协作变得…

Java面试题(高频、有答案,全网最强)

原文网址&#xff1a;Java面试题&#xff08;高频、有答案&#xff0c;全网最强&#xff09;-CSDN博客 这是一套全网最强的Java面试题&#xff0c;吊打网上所有Java面试题。 此套面试题的威力&#xff1a;看过这套题的朋友、同事、粉丝参加了面试后说&#xff0c;他们面试被问…

基于C#的GRPC

GRPC gRPC&#xff08;gRPC Remote Procedure Call&#xff09;是由Google开发的高性能、跨语言的远程过程调用框架。它基于HTTP/2协议进行通信&#xff0c;支持多种编程语言&#xff0c;包括C, C#, Java, Python等&#xff0c;使不同语言的应用程序可以通过远程调用相互通信。…

SPASS教程-入门

常用的统计工具 EXCEL 严格说来并不是统计软件&#xff0c;但作为数据表格软件&#xff0c;有一定统计计算功能。对于简单分析&#xff0c;Excel还算方便&#xff0c;但随着问题的深入&#xff0c;Excel就不那么“傻瓜”&#xff0c;需要使用函数&#xff0c;甚至根本没有相应…

​软考-高级-信息系统项目管理师教程 第四版【第19章-配置与变更管理-思维导图】​

软考-高级-信息系统项目管理师教程 第四版【第19章-配置与变更管理-思维导图】 课本里章节里所有蓝色字体的思维导图

Spire.Office for Java 8.10.2 同步更新Crk

Spire.Office for Java 是 E-iceblue 提供的企业级 Office Java API 的组合。它包括Spire.Doc for Java、Spire.XLS for Java、Spire.Presentation for Java、Spire.PDF for Java和Spire.Barcode for Java。 开发人员可以使用Spire.Office for Java在Java应用程序中执行各种办…

【electron】【附排查清单】记录一次逆向过程中,fetch无法请求http的疑难杂症(net::ERR_BLOCKED_BY_CLIENT)

▒ 目录 ▒ &#x1f6eb; 导读需求开发环境 1️⃣ Adblock等插件拦截2️⃣ 【失败】Content-Security-Policy启动服务器json-serverhtml中的meta字段 3️⃣ 【失败】https vs httpwebPreferences & allowRunningInsecureContent disable-features 4️⃣ 【失败】检测fetch…

技术分享 | app自动化测试(Android)--元素定位方式与隐式等待

元素定位是 UI 自动化测试中最关键的一步&#xff0c;假如没有定位到元素&#xff0c;也就无法完成对页面的操作。那么在页面中如何定位到想要的元素&#xff0c;本小节讨论 Appium 元素定位方式。 Appium的元素定位方式 定位页面的元素有很多方式&#xff0c;比如可以通过 I…

初识Java 17-2 反射

目录 转型前检查 构建例子&#xff1a;生成层次结构 优化Creator&#xff1a;使用类字面量 优化PetCounter&#xff1a;动态验证类型 更通用的递归计数 注册工厂 本笔记参考自&#xff1a; 《On Java 中文版》 转型前检查 当我们使用传统的类型转换&#xff0c;例如&…

实战!工作中常用的设计模式

文章目录 前言一、策略模式1.1、 业务场景1.2 、策略模式定义1.3、 策略模式使用1.3.1、一个接口&#xff0c;两个方法1.3.2、不同策略的差异化实现1.3.3、使用策略模式 二、责任链模式2.1、业务场景2.2、责任链模式定义2.3、责任链模式使用2.3.1、一个接口或者抽象类2.3.2、每…

11.7加减计数器,可置位~,数字钟分秒,串转并,串累加转并,24位串并128,流水乘法器,一些乘法器

信号发生器 方波&#xff0c;就是一段时间内都输出相同的信号 锯齿波就是递增 三角波就是先增后减 加减计数器 当mode为1则加&#xff0c;Mode为0则减&#xff1b;只要为0就输出zero 这样会出问题&#xff0c;因为要求是十进制&#xff0c;但是这里并没有考虑到9之后怎么办&a…

openvino学习(一)ubuntu20.04安装openvino2022

安装openvino2022要求 操作系统 Ubuntu 18.04 长期支持 (LTS)&#xff0c;64 位 Ubuntu 20.04 长期支持 (LTS)&#xff0c;64 位 软件 CMake 3.13 或更高版本&#xff0c;64 位 GCC 7.5.0&#xff08;适用于 Ubuntu 18.04&#xff09;或 GCC 9.3.0&#xff08;适用于 Ubunt…

工具介绍——第三方软件远程连接(工具:Rustdesk)

文章目录 前言一、使用工具二、开始演示1、拿下目标主机权限后上传文件2、运行目标主机上的rustdesk-1.1.9.exe文件3、目标主机上whoami查看现在的用户4、查找目标主机上连接的文件&#xff0c;并添加连接密码5、目标主机重启rustdesk的应用程序6、本地连接主机 前言 这里主要…

“第六十三天”

这两天怎么做的这么别扭&#xff0c;为什么我的vs 的strlen函数包括终止字符了&#xff1b; 哦哦&#xff0c;明白了&#xff0c;fgets函数读取在未达到指定字长&#xff0c;或者遇见空白符之前&#xff0c;会读取前面的所有字符&#xff0c;所以会读取换行符&#xff0c;而get…

康耐视深度学习ViDi-View菜单介绍

Accept View承认当前图片标注的有效性 Clear Marking 清除当前图片的标注特征 Clear Marking & Labels清除当前图片的标注特征和标签 Process处理当前图片 Edit ROI编辑检测的区域 Edit Regions编辑(标注)特征区域 Edit Mask 编辑遮挡(屏蔽)区域 Apply Mask To Tool将遮挡(…

杨辉三角(Java实现)

public class Demo09 {public static void main(String[] args) {//第一位和最后一位都为1//除了每行的第一列之外,其余的数为a[i][j]aa[i-1][j]a[i-1][j-1];int[][] arrays new int[10][10];for(int i 0;i<10;i){if(i0){ //第一行直接赋值为0arrays[i][0]1;continue;//重…

深入理解强化学习——多臂赌博机:基于置信度上界的动作选择

分类目录&#xff1a;《深入理解强化学习》总目录 因为对动作—价值的估计总会存在不确定性&#xff0c;所以试探是必须的。贪心动作虽然在当前时刻看起来最好&#xff0c;但实际上其他一些动作可能从长远看更好。 ϵ − \epsilon- ϵ−贪心算法会尝试选择非贪心的动作&#xf…

ZipInputStream解压报错java.lang.IllegalArgumentException: MALFORMED

背景 使用jdk自带的zip工具ZipInputStream&#xff0c;去读取使用winrar在中文windows制作出来的zip文件报错。 Exception in thread "main" java.lang.IllegalArgumentException: MALFORMEDat java.util.zip.ZipCoder.toString(ZipCoder.java:58)at java.util.zip…