ZeroMQ(java)之Push/Pull模式

在ZeroMQ中并没有绝对的服务端与客户端之分，所有的数据接收与发送都是以连接为单位的，只区分ZeroMQ定义的类型，例如Response与Request，Publisher与Subscriber，Push与Pull等。。。

例如在前面我们最开始的Response/Request模式，因为只有一个Response端，而有多个Request端，所以我们选择在Response端调用bind方法来建立监听，而在Request端调用connect方法与Response端建立连接。。。因此根据以前常用的概念，可以简单的将Response理解为服务端，将Request理解为客户端。。。。

这种状态下，整个系统大概用下面的图形来形容：



这里因此按照我们常规的区分方法，将建立监听的叫做服务端，发起连接的叫做客户端，但其实呢，在ZeroMQ这种按照监听的方式来区分是不成立的。。。将上图的网络构建变成如下这个样子：



上图的网络结构中，在Request端建立监听，而在Response发起与Request端的连接，这样，request同样可以发送请求到Response端。。。

其实说这么多无非就像强调：在ZeroMQ中，不要用常规的server/client模式来对组件进行分类，而应该采用ZeroMQ中定义的类型（Request，Response，Push，Pull）。

好了，接下来回到Push/Pull模式，这算是非常经典的了吧，Push产生消息，Pull角色来拿消息。。。甚至可以用生产者/消费者模型来对应。。。应用场景就非常的广泛了。。。最典型的应用场景就是任务分发。。。

在ZeroMQ中根据Push与Pull角色各自的数量又定义了一些比较有趣的名词：

（1）parallel pipeline，并行流水线，这种情况下是一个push，多个pull，可以理解为一个push不断的产生任务，并将这些任务分发给pull角色。。。如下图：



在这种网络结构中，manager不断的将任务分发给worker，要实现这种通信，在ZeroMQ中就直接用Push/Pull就可以了，代码也很简单，首先是Manager端（Push）：

[java] view
plaincopy





package pushpull13;  

  

import org.zeromq.ZMQ;  

  

public class Push {  

    public static void main(String args[]) {  

  

        ZMQ.Context context = ZMQ.context(1);  

        ZMQ.Socket push  = context.socket(ZMQ.PUSH);  

        push.bind("ipc://fjs");  

  

        for (int i = 0; i < 10000000; i++) {  

            push.send("hello".getBytes());  

        }  

        push.close();  

        context.term();  

      

    }  

}  

代码很简单吧，这里建立的是PUSH类型的socket，然后循环一千万次，给建立连接的worker发送数据，那么接下来来看看Worker（Pull）部分的代码：

[java] view
plaincopy





package pushpull13;  

  

import java.util.concurrent.atomic.AtomicInteger;  

  

import org.zeromq.ZMQ;  

  

public class Pull {  

    public static void main(String args[]) {  

        final AtomicInteger number = new AtomicInteger(0);  

        for (int i = 0; i < 5; i++) {  

            new Thread(new Runnable(){  

                private int here = 0;  

                public void run() {  

                    // TODO Auto-generated method stub  

                    ZMQ.Context context = ZMQ.context(1);  

                    ZMQ.Socket pull = context.socket(ZMQ.PULL);  

                    pull.connect("ipc://fjs");  

                    //pull.connect("ipc://fjs");  

                    while (true) {  

                        String message = new String(pull.recv());  

                        int now = number.incrementAndGet();  

                        here++;  

                        if (now % 1000000 == 0) {  

                            System.out.println(now + "  here is : " + here);  

                        }  

                    }  

                }  

                  

            }).start();  

              

        }  

    }  

}  

这里建立了5个worker，建立于与manager的连接，这里可能就会涉及到一个问题，manager是怎么将数据分发给这5个woker的呢，这里由于还没有看过实现代码，所以不知道这里具体是怎么个策略，不过后来测试数据之后发现各个worker之间收到的数量相差不大，可以猜测大概是轮询发送的。。。。

好了，到这里整个所谓的并行流水线的网络构建就算是差不多了。。。那么接下来来看另外一种，这里Push与Pull之间的对应关系是多个Push角色对应一个Pull角色，在ZeroMQ中，给这种结构取的名叫做公平队列，结构如下图：



这里也就是说将Pull角色理解为一个队列，各个Push角色不断的向这个队列中发送数据。。。这种结构应用场景也很广泛吧，例如分布式的数据统计啥的。。。

好了，来看看实现代码形式吧，先来看看Push：

[java] view
plaincopy





package pushpull31;  

  

import org.zeromq.ZMQ;  

  

public class Push {  

    public static void main(String args[]) {  

        for (int j = 0; j < 3; j++) {  

            new Thread(new Runnable(){  

  

                public void run() {  

                    // TODO Auto-generated method stub  

                    ZMQ.Context context = ZMQ.context(1);  

                    ZMQ.Socket push = context.socket(ZMQ.PUSH);  

                      

                    push.connect("ipc://fjs");  

                      

                    for (int i = 0; i < 10000000; i++) {  

                        push.send("hello".getBytes());  

                        System.out.println(i);  

                    }  

                    push.close();  

                    context.term();  

                }  

                  

            }).start();;  

        }  

    }  

}  

由于这部分push与pull的关系是多对一，所以选择在pull端建立监听，让push端来连接pull端。。。代码还是很简单的吧。。。

不过这里有一个比较有意思的现象，加入我们先运行push端，而这个时候pull端并没有运行的话，会发现send方法也会被执行，只不过执行一会以后就阻塞了，这里可以猜测，ZeroMQ的push端是先将数据写到了一个缓冲区，然后数据是从缓冲区中写到已经建立好连接的pull端的，当然这个只是猜测，具体是什么样子的以后看看源码的实现就知道了。。。。

好了，接下来来看看Pull端的代码实现吧：

[java] view
plaincopy





package pushpull31;  

  

import org.zeromq.ZMQ;  

  

public class Pull {  

    public static void main(String args[]) {  

        ZMQ.Context context = ZMQ.context(1);  

        ZMQ.Socket pull = context.socket(ZMQ.PULL);  

          

        pull.bind("ipc://fjs");  

          

        int number = 0;  

        while (true) {  

            String message = new String(pull.recv());  

            number++;  

            if (number % 1000000 == 0) {  

                System.out.println(number);  

            }  

        }  

    }  

}  

这里够简单吧，建立一个连接，然后循环里不断的接收数据就好了。。。

好了，到这里Push/Pull中的一对多与多对一的网络结构就算是讲完了吧，其实还有一种多对多的结构。。不过意思都差不多吧。。。。。就不具体弄出来了。。