开发者社区> java牛> 正文

ZeroMQ(java)中的数据流SessionBase与SocketBase

简介: 前面的文章中已经比较的清楚了ZeroMQ(java)中如何在底层处理IO, 通过StreamEngine对象来维护SelectableChannel对象以及IO的事件回调,然后通过Poller对象来维护Selector对象,然后用IOObject对象来具体的管理SelectableChannel对...
+关注继续查看

前面的文章中已经比较的清楚了ZeroMQ(java)中如何在底层处理IO,

通过StreamEngine对象来维护SelectableChannel对象以及IO的事件回调,然后通过Poller对象来维护Selector对象,然后用IOObject对象来具体的管理SelectableChannel对象在Poller上面的注册,以及事件回调,他们之间的关系可以用下面的图形来简单的描述一下:



对于接收到的数据,首先由StreamEngine进行处理,其实它会调用内部的decoder将字节数据转化为Msg对象,然后再交给上层的对象,

其实这里的上层对象也就是Session对象,每一个StreamEngine对象都有一个自己的Session对象,然后Session对象收到下面传上来的数据之后,再会通过Pipe,将数据发送到其更上层的Socket对象,

然后接下来的数据处理就交由用户的代码来处理了。。。

对于刚刚提到的对象之间的层次,用下面的图形来描述吧:


这张图应该还算刻画的比较直接了吧,底层数据通信部分负责从channel接收数据和发送二进制的数据,然后又Decoder以及Encoder来负责字节数据与Msg之间的转化。。。。

我们在ZMQ中会看到很多种类的Socket,例如Req,Dealer,Router啥的,他们都继承自SocketBase类型,每一种类型都有自己的Session,都继承自SessionBase类型。。。

好了,那么接下来先来看看SessionBase类型吧,每一个StreamEngine对象都有一个Session对象与之关联,他们是一对一的关系,先来看看它的一些重要的属性定义吧:

[java] view plaincopy在CODE上查看代码片派生到我的代码片
  1. private boolean connect;  //是否需要连接,如果是fasle的话,那么表示是listener创建的连接  
  2.   
  3. private Pipe pipe;   //与socket进行通信的pipe  
  4. private final Set<Pipe> terminating_pipes;  
  5. //如果是true的话,表示还有message在pipe里面没有执行  
  6. private boolean incomplete_in;  
  7.   
  8. //如果是true的话,表示停止发送数据到网络中  
  9. private boolean pending;  
  10.   
  11. private IEngine engine;  //绑定到这个session上面的底层的通信  
  12.   
  13. private SocketBase socket;   //当前session所属的socket  
  14.   
  15.   
  16. private IOThread io_thread;    //当前session关联的IO线程,底层的io将会加入到这个IO线程  
  17. private static int linger_timer_id = 0x20;  
  18.   
  19. //  True is linger timer is running.  
  20. private boolean has_linger_timer;    //有超时事件的注册?  
  21. private boolean identity_sent;    //如果是true的话,表示标志已经发送了  
  22. private boolean identity_received;  //表示标志已经接收了  
  23. private final Address addr;   //连接的地址  
  24.   
  25. private IOObject io_object;   //关联的IO对象,IEngine里的io对象  

具体这些属性的是干嘛的,上面的注释基本上都已经给出来的吧,这里比较重要的属性是:

pipe,它用于与上面的Socket进行通信,当下层有数据被解析出来以后,会通过pipe将msg发送给上层的socket,具体pipe的过程,前面的文章已经说过了。。。

IEngine,底层数据通信的StreamEngine对象的引用,这个重要性就不说了吧,。。。

IOThread对象,这个是当前Session对象将会依赖的IO线程,也就是发给session的命令都会被这个IO线程的mailbox接收到,从而在这个线程中执行命令,嗯。。。重要吧。。。

另外还有一些标志位什么的。。。

好了,这里就不细说,来看看一些重要的方法吧:

[java] view plaincopy在CODE上查看代码片派生到我的代码片
  1. //与pipe关联上,这里其实主要是为了将当前pipe的事件回调设置为当前对象  
  2. public void attach_pipe(Pipe pipe_) {  
  3.     assert (!is_terminating ());  
  4.     assert (pipe == null);  
  5.     assert (pipe_ != null);  
  6.     pipe = pipe_;  //保存当前的pipe  
  7.     pipe.set_event_sink (this);  //将当前的pipe的事件回调社设置为当前  
  8. }  

用于关联pipe对象,这里可以看到将pipe的事件回到设置成了当前session对象。。那么来看看这些事件回调方法是怎么处理的吧:

[java] view plaincopy在CODE上查看代码片派生到我的代码片
  1. //当有数据传给当前的pipe的时候,其实也就是pipe的对面socket那边发送数据给这里了,然后发送pipe可读的命令,那么表示有数据需要通过底层的engine发送出去了  
  2. public void read_activated(Pipe pipe_)  {  
  3.     // Skip activating if we're detaching this pipe  
  4.     if (pipe != pipe_) {  
  5.         assert (terminating_pipes.contains (pipe_));  
  6.         return;  
  7.     }  
  8.       
  9.     if (engine != null)  
  10.         engine.activate_out ();  //激活底层engine的channel的写事件  
  11.     else  
  12.         pipe.check_read ();  
  13. }  
  14.   
  15. //表示可以发送数据到pipe了,那么表示需要到底层的engine接收数据了  
  16. public void write_activated (Pipe pipe_)  {  
  17.     // Skip activating if we're detaching this pipe  
  18.     if (pipe != pipe_) {  
  19.         assert (terminating_pipes.contains (pipe_));  
  20.         return;  
  21.     }  
  22.   
  23.   
  24.     if (engine != null)  
  25.         engine.activate_in ();  //激活底层的engin的channel上面的读取事件,也就是通知底层的engin应该从网络接收数据了  
  26. }  

这里先是pipe可以读的时候的事件回调方法,这个处理很简单吧,直接激活底层StreamEngine的channle在poller上注册写事件,那么当底层channel可以写数据的时候,就会从当前session的pipe里去读取数据,然后发送出去。。

第二个方法是当pipe可以写的时候,这个其实就是直接注册channel的读取事件,那么当channel就会去接收数据,最后这些他们都会通过pipe发送给上层的socket对象。。。到此上面那张图的整个运行情况应该都很清楚了吧。。。

那么Session的最为关键的地方也就差不多了。。。还有一些细节,以后有需要的话再介绍吧。。。

好了接下来来看看SocketBase这个类型吧,前面已经说到了ZMQ中所有的Socket类型多继承自这个类型,可见他的重要性。。。先来看看它的一些重要的属性定义吧:

[java] view plaincopy在CODE上查看代码片派生到我的代码片
  1. private final Map<String, Own> endpoints;  //这里保存所有打开的endpoint,这里key是连接地址,value其实是Session对象  
  2. private final Map<String, Pipe> inprocs;   //IPC通信方法  
  3.   
  4. //  Used to check whether the object is a socket.  
  5. private int tag;   //标志位,用于判断当前对象是否是socket类型的  
  6.   
  7. private boolean ctx_terminated;    //如果是true,表示关联的context已经停止了  
  8.   
  9. private boolean destroyed;  //如果是true的话,表示当前socket对象已经被命令销毁了  
  10. private final Mailbox mailbox;   //邮箱,用于接收别的地方发送过来的命令  
  11. private final List<Pipe> pipes;   //当前所有关联的pipe,这些pipe是用于与当前socket的所有session进行数据通信的  
  12. private Poller poller;  //poller对象  
  13. private SelectableChannel handle;   //这个是mailbox的handler  
  14. private long last_tsc;   //上一次执行命令的时间  
  15. private int ticks;    //上次执行完命令之后,收到的message  
  16. private boolean rcvmore;  //接着还有message要接收  
  17. private SocketBase monitor_socket;  //用于监控的socket  
  18. private int monitor_events;  
  19. protected ValueReference<Integer> errno;  

这里比较重要的有,:

enpoints,用于保存所有的session与其连接地址的,

pipes,保存所有与底层的session关联的pipe,这里Socket与Sesssion之间的关系是一对多的...

mailbox,socket也有自己的mailbox,不用依附于IOThread对象,不过这里有个坑,Socket类型的对象有自己的mailbox,不用依附于IO线程,并不意味着它就有自己的线程,因为它直接依赖于用户线程,依赖于用户代码...也就是说mailbox里面的命令的执行都是在用户线程中搞定的...呵呵,刚开始这个地方还纠结了很久...

来看看它的pipe的事件回调吧:

[java] view plaincopy在CODE上查看代码片派生到我的代码片
  1.     //当pipe可以读的时候需要执行的回调方法,表示底层的StreamEngin有数据发送到这里了  
  2.     public void read_activated (Pipe pipe_)  {  
  3.         xread_activated(pipe_);  //调用子类的方法来具体处理这些数据  
  4.     }  
  5.   
  6. //当pipe可以写的时候执行的回调  
  7.     public void write_activated (Pipe pipe_) {  
  8.         xwrite_activated (pipe_);  
  9.     }  

其实这里没有太多的内容,因为都是在具体的子类中完成的,不过到这里整个ZeroMQ(java)中数据是怎么进行流动的就算已经很清楚了...当然,有一些细节性的东西还没有列出来..


好了,到现在为止,ZeroMQ(java)中就还剩下具体的socket类型的运行以及编码方式两个大的地方没有分析了...

若转载请注明出处!若有疑问,请回复交流!

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
阿里云服务器怎么设置密码?怎么停机?怎么重启服务器?
如果在创建实例时没有设置密码,或者密码丢失,您可以在控制台上重新设置实例的登录密码。本文仅描述如何在 ECS 管理控制台上修改实例登录密码。
19696 0
ML之xgboost:利用xgboost算法(自带方式)训练mushroom蘑菇数据集(22+1,6513+1611)来预测蘑菇是否毒性(二分类预测)
ML之xgboost:利用xgboost算法(自带方式)训练mushroom蘑菇数据集(22+1,6513+1611)来预测蘑菇是否毒性(二分类预测)
36 0
阿里云服务器端口号设置
阿里云服务器初级使用者可能面临的问题之一. 使用tomcat或者其他服务器软件设置端口号后,比如 一些不是默认的, mysql的 3306, mssql的1433,有时候打不开网页, 原因是没有在ecs安全组去设置这个端口号. 解决: 点击ecs下网络和安全下的安全组 在弹出的安全组中,如果没有就新建安全组,然后点击配置规则 最后如上图点击添加...或快速创建.   have fun!  将编程看作是一门艺术,而不单单是个技术。
17986 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,阿里云优惠总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系.
24792 0
Oracle 数据库利用sql语句杀掉用户session进程,“ORA-01940: 无法删除当前连接的用户“问题解决办法
Oracle 数据库利用sql语句杀掉用户session进程,“ORA-01940: 无法删除当前连接的用户“问题解决办法
40 0
Chrome 浏览器降级后浏览网站不保留用户数据问题原因及解决方法
Chrome 浏览器降级后浏览网站不保留用户数据问题原因及解决方法
58 0
ML之xgboost:利用xgboost算法(sklearn+7CrVa)训练mushroom蘑菇数据集(22+1,6513+1611)来预测蘑菇是否毒性(二分类预测)
ML之xgboost:利用xgboost算法(sklearn+7CrVa)训练mushroom蘑菇数据集(22+1,6513+1611)来预测蘑菇是否毒性(二分类预测)
50 0
+关注
java牛
善于分享,乐于分享
281
文章
2
问答
文章排行榜
最热
最新
相关电子书
更多
OceanBase 入门到实战教程
立即下载
阿里云图数据库GDB,加速开启“图智”未来.ppt
立即下载
实时数仓Hologres技术实战一本通2.0版(下)
立即下载