ZeroMQ(java)中组件间数据传输（Pipe的实现）

在ZeroMQ（java）中，整个IO的处理流程都是分层来进行的，当然处于最下端的肯定是前面介绍过的poller以及StreamEngin了。。。。涉及到上层的话就还有session，以及socket，先用一张图来大概的描述一下整个层次关系吧。。

整个分层的结构大概就是这样吧，其中poller与StreamEngin是怎么交互的，这个就不说饿了吧，然后Session这个怎么与session之间交互呢，这个以后再说吧，其实在streamEngin里面有自己的session引用。。反正这里没啥意思。。主要就在与Session怎么与自己所属的Socket进行交互，当从最底层接收到数据之后，session如何交给上层的socket，让其来处理。。。这里就涉及到了Pipe，也就是session与自己所属的socket之间是通过pipe来进行数据传递的。。。

那么在具体的分析session与socket之前就来看看这个Pipe是怎么工作的吧，先来大概的看看它的类图：

这里可以看到Pipe继承自ZObject类型，那么可以知道Pipe可以发送，接受以及执行命令，同时也就意味着Pipe也需要由自己关联的IO线程才行，或者说有关联的mailbox。。。不过这个也不是强制的，以后再分析Socket的pipe的时候，就会发现它的pipe关联到socket自己的mailbox，但是socket的mailbox没有注册到任何的poller上面去，也就是它并没有在任何IO线程里执行，最后其实是在用户代码的线程中运行的。。。。好了。好像闲话说的比较多了。。用一张图来刻画一下Pipe是怎么运行的吧：

其实通过这张图形就已经将Pipe的运行原理基本描述出来了，pipe的两端都分别关联了两个YPipe（可以将其理解为队列）对象，例如左边将其中一个YPipe当做写端，那么在另外一边就将其看成是读端。。。

这里的YPipe对象可以将其理解为队列，至于说具体的实现，底层确实是队列，只不过是自己实现的，而且实现的还挺繁琐的，就不细说了，不过这里有向吐槽的地方，明明concurrent库中有无锁的队列ConcurrentLinkedList，在并发环境下有很好的性能，干嘛不在这个基础上进行扩展。。。。

这里另外还要看看在ZeroMQ中，也定义的有Pipe类型自己的事件回调，其定义如下：

[java] view plain copy

public interface IPipeEvents {
void read_activated(Pipe pipe); //有数据可以读取
void write_activated(Pipe pipe); //当前pipe有数据写
void hiccuped(Pipe pipe); //对面的pipe替换掉了读端，也就是当前需要替换写段的时候的回调
void terminated(Pipe pipe); //当前pipe停止的回调
}

具体每个方法是干嘛用的注释应该说的很清楚了。。那么接下来来看看Pipe的两端是怎么进行交互的吧，首先看如何发送数据到pipe的另外一端：

[java] view plain copy

//从写端写数据局，发送给pipe的另外一端
public boolean write (Msg msg_) {
if (!check_write ())
return false;
boolean more = msg_.has_more();
outpipe.write (msg_, more);
if (!more)
msgs_written++; //已经读取的msg的计数
return true;
}

其实这里直接就是在写端，将数据写到队列里面去就好了，那么如何通知对面当前有数据发送过来了呢，来看另外一个方法：

[java] view plain copy

//其实这里主要是给对面的pipe发送activate_read命令，表示它可以读了
public void flush () {
// The peer does not exist anymore at this point.
if (state == State.terminating)
return;
if (outpipe != null && !outpipe.flush ()) {
send_activate_read (peer); //向对面发送可以读取的命令
}
}

这个，如果看了ZObject就应该很清楚了吧，直接给命令的另外一端发送activate_read类型的命令，那么这个命令最终将会被pipe的另外一端所关联的mailbox收到，从而对面的Pipe将会在其IO线程中执行命令，对于这个命令，进行的操作是process_activate_read方法，那么来看看Pipe中这个方法的的定义吧：

[java] view plain copy

//收到命令，表示底层的pipe有数据可以读取了，这里主要是要调用事件回调，通知上层的代码，pipe有数据可以读取了
protected void process_activate_read () {
if (!in_active && (state == State.active || state == State.pending)) {
in_active = true;
sink.read_activated (this); //调用事件回调
}
}

这里其实就是调用当前的pipe的事件回调，来处理当前的pipe对象，其实也就是通知上层的代码，当前pipe有数据可以读了，让其进行处理。。。。

好了，那么到这里整个Pipe的运行原理就算比较的清楚了。。。

不过自己不太明白，在java中这种数据的传递明明很简单就可以实现，干嘛要搞的这么复杂。。。不过这里也有一个好处，就是将每一个对象的方法的执行都封闭在了自己的IO线程内部。。。也算是一种线程封闭原则的实现吧。。。其余的好处，好像没啥好处，而且真的觉得略繁琐。。。。

巴特西

ZeroMQ(java)中组件间数据传输（Pipe的实现）

最新文章

热门文章