博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
ZeroMQ之Push与Pull (Java)
阅读量:5844 次
发布时间:2019-06-18

本文共 3630 字,大约阅读时间需要 12 分钟。

本系列文章均转自:http://blog.csdn.net/kobejayandy/article/details/20163431

在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] 
  1. package pushpull13;  
  2.   
  3. import org.zeromq.ZMQ;  
  4.   
  5. public class Push {  
  6.     public static void main(String args[]) {  
  7.   
  8.         ZMQ.Context context = ZMQ.context(1);  
  9.         ZMQ.Socket push  = context.socket(ZMQ.PUSH);  
  10.         push.bind("ipc://fjs");  
  11.   
  12.         for (int i = 0; i < 10000000; i++) {  
  13.             push.send("hello".getBytes());  
  14.         }  
  15.         push.close();  
  16.         context.term();  
  17.       
  18.     }  
  19. }  

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

[java] 
  1. package pushpull13;  
  2.   
  3. import java.util.concurrent.atomic.AtomicInteger;  
  4.   
  5. import org.zeromq.ZMQ;  
  6.   
  7. public class Pull {  
  8.     public static void main(String args[]) {  
  9.         final AtomicInteger number = new AtomicInteger(0);  
  10.         for (int i = 0; i < 5; i++) {  
  11.             new Thread(new Runnable(){  
  12.                 private int here = 0;  
  13.                 public void run() {  
  14.                     // TODO Auto-generated method stub  
  15.                     ZMQ.Context context = ZMQ.context(1);  
  16.                     ZMQ.Socket pull = context.socket(ZMQ.PULL);  
  17.                     pull.connect("ipc://fjs");  
  18.                     //pull.connect("ipc://fjs");  
  19.                     while (true) {  
  20.                         String message = new String(pull.recv());  
  21.                         int now = number.incrementAndGet();  
  22.                         here++;  
  23.                         if (now % 1000000 == 0) {  
  24.                             System.out.println(now + "  here is : " + here);  
  25.                         }  
  26.                     }  
  27.                 }  
  28.                   
  29.             }).start();  
  30.               
  31.         }  
  32.     }  
  33. }  

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

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

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

好了,来看看实现代码形式吧,先来看看Push:

[java] 
  1. package pushpull31;  
  2.   
  3. import org.zeromq.ZMQ;  
  4.   
  5. public class Push {  
  6.     public static void main(String args[]) {  
  7.         for (int j = 0; j < 3; j++) {  
  8.             new Thread(new Runnable(){  
  9.   
  10.                 public void run() {  
  11.                     // TODO Auto-generated method stub  
  12.                     ZMQ.Context context = ZMQ.context(1);  
  13.                     ZMQ.Socket push = context.socket(ZMQ.PUSH);  
  14.                       
  15.                     push.connect("ipc://fjs");  
  16.                       
  17.                     for (int i = 0; i < 10000000; i++) {  
  18.                         push.send("hello".getBytes());  
  19.                         System.out.println(i);  
  20.                     }  
  21.                     push.close();  
  22.                     context.term();  
  23.                 }  
  24.                   
  25.             }).start();;  
  26.         }  
  27.     }  
  28. }  

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

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

好了,接下来来看看Pull端的代码实现吧:

[java] 
  1. package pushpull31;  
  2.   
  3. import org.zeromq.ZMQ;  
  4.   
  5. public class Pull {  
  6.     public static void main(String args[]) {  
  7.         ZMQ.Context context = ZMQ.context(1);  
  8.         ZMQ.Socket pull = context.socket(ZMQ.PULL);  
  9.           
  10.         pull.bind("ipc://fjs");  
  11.           
  12.         int number = 0;  
  13.         while (true) {  
  14.             String message = new String(pull.recv());  
  15.             number++;  
  16.             if (number % 1000000 == 0) {  
  17.                 System.out.println(number);  
  18.             }  
  19.         }  
  20.     }  
  21. }  

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

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

你可能感兴趣的文章
FreeMarker表达式
查看>>
Debian9.2 下使用vnstat查看服务器带宽流量统计
查看>>
NGINX + PHP-FPM 502
查看>>
Openstack API常用命令
查看>>
OpenSSL漏洞凶猛来袭 慧眼恶意代码监测应对有方
查看>>
C语言 喝汽水问题
查看>>
ubuntu 下安装 mysql
查看>>
关于k-means聚类算法的matlab实现
查看>>
一键安装Gitlab后的备份、迁移与恢复
查看>>
因为本人工作繁忙,精力有限,本博客停止更新。有兴趣的博友可以关注我在CSDN上的主博客...
查看>>
SQL server查看触发器是否被禁用
查看>>
跟随我在oracle学习php(8)
查看>>
UVA-10212 The Last Non-zero Digit. 分解质因子+容斥定理
查看>>
求两个集合的交集,并集,差集
查看>>
Kotlin的语法糖(一)基础篇
查看>>
OkHttp源码分析
查看>>
NS_OPTIONS枚举的用法
查看>>
QAQ高精度模板笔记√
查看>>
【Android笔记】入门篇02:全屏设置和禁止横屏竖屏切换
查看>>
Kubernetes的本质
查看>>