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

前面的文章中已经比较的清楚了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

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

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

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

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

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

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

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

[java] view plaincopy

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

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. if (engine != null)
9. engine.activate_out ();  //激活底层engine的channel的写事件
10. else
11. pipe.check_read ();
12. }
13. //表示可以发送数据到pipe了，那么表示需要到底层的engine接收数据了
14. public void write_activated (Pipe pipe_)  {
15. // Skip activating if we‘re detaching this pipe
16. if (pipe != pipe_) {
17. assert (terminating_pipes.contains (pipe_));
18. return;
19. }
20. if (engine != null)
21. engine.activate_in ();  //激活底层的engin的channel上面的读取事件，也就是通知底层的engin应该从网络接收数据了
22. }

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

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

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

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

[java] view plaincopy

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

这里比较重要的有,:

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

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

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

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

[java] view plaincopy

1. //当pipe可以读的时候需要执行的回调方法，表示底层的StreamEngin有数据发送到这里了
2. public void read_activated (Pipe pipe_)  {
3. xread_activated(pipe_);  //调用子类的方法来具体处理这些数据
4. }
5. //当pipe可以写的时候执行的回调
6. public void write_activated (Pipe pipe_) {
7. xwrite_activated (pipe_);
8. }

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

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