《ZeroMQ介绍》由会员分享，可在线阅读，更多相关《ZeroMQ介绍（5页珍藏版）》请在金锄头文库上搜索。
1、ZeroMQ 介绍1介绍 MQ 有很多种，比如 MSMQ，ActiveMQ，RabbitMQ 等等 今天主要介绍的 MQ 是 ZeroMQ，其具有一个独特的非中间件的模式，不需要安装和 运行一个消息服务器，或中间件。你只需要简单的引用 ZeroMQ 程序库，然后你就可以愉快 的在应用程序之间发送消息了，因此部署是非常简单的。就性能上来说比其他几个 MQ 要强 的非常多。 ZeroMQ，新一层的网络堆栈，轻量级，socket 风格的接口，支持多种消息传送模式， 如：P2P，订阅发布模式以及请求-响应模式等等，支持多种底层协议，支持多种平台和 CPU 架构。2模型服务端：public class
2、Server public static void main(String args) throws Exception / Prepare our context and socketsfinal ZMQ.Context context = ZMQ.context(1);ZMQ.Socket socket = context.socket(ZMQ.REP);socket.bind(“tcp:/*:5555“); while (true) byte request = socket.recv(0);System.out.println(new String(request);while (so
3、cket.hasReceiveMore() request = socket.recv(0);System.out.println(new String(request);String s = new String(request);long r = Long.parseLong(s) + 100;/ reply = request+100socket.send(String.valueOf(r).getBytes(), 0);客户端：public class Client public static void main(String args) / Prepare our context a
4、nd socketfinal ZMQ.Context context = ZMQ.context(1);ZMQ.Socket socket = context.socket(ZMQ.REQ); System.out.println(“Connecting to hello world server.“);socket.connect(“tcp:/localhost:5555“); while (true) long l = System.nanoTime();String s = String.valueOf(l);socket.send(s.getBytes(), 0);byte r = s
5、ocket.recv(0);System.out.println(l + “ + 100 = “ + new String(r);3设计模式ZeroMQ 把通讯的需求看成四类。其中一类是一对一结对通讯，用来支持传统的 TCP socket 模型，但并不推荐使用。常用的通讯模式只有三类1.请求回应模型。由请求端发起请求，并等待回应端回应请求。从请求端来看，一定是一对收发配对的；反之，在回应端一定是发收对。请求端和回应端都可以是 1:N 的模型。通常把 1 认为是 server ，N 认为是 Client 。ZeroMQ 可以很好的支持路由功能（实现路由功能的组件叫作 Device），把 1:N
6、扩展为 N:M （只需要加入若干路由节点）。从这个模型看，更底层的端点地址是对上层隐藏的。每个请求都隐含有回应地址，而应用则不关心它。2.发布订阅模型。这个模型里，发布端是单向只发送数据的，且不关心是否把全部的信息都发送给订阅端。如果发布端开始发布信息的时候，订阅端尚未连接上来，这些信息直接丢弃。不过一旦订阅端连接上来，中间会保证没有信息丢失。同样，订阅端则只负责接收，而不能反馈。如果发布端和订阅端需要交互（比如要确认订阅者是否已经连接上），则使用额外的 socket 采用请求回应模型满足这个需求。3.管道模型。这个模型里，管道是单向的，从 PUSH 端单向的向 PULL 端单向的推送数据流。
7、如果有多个 client 端同时连接到这个 server,则服务器会在内部做一个负载均衡,采用平均分配的算法,将所有的消息均衡发布到 client 端上2,3 两种模式中,无论是 server 端还是 client 端在启动时启动时 都是不知道 client 实际数 量的,这就意味着,一个使用 zeromq 搭建的服务,可以进行”热更新”.1) request-reply 就是一般的 C/S 架构中,client 与 server 之间一问一答的通信模式,比如最经典的 echo 服务. 需要注意的是,client 发送一个 request,server 必须有一个回应. 这个 pattern
8、并不是什么亮点,下面亮点来了. 2) publish-subscribe server 端作为 publish 端,而任何连接到服务端的 client 都会成为 subscribe 端.也就是说,server 端会把当前的所有需要 publish 出去的消息全部发送到当前连接上去的 client 上.3) push-pull server 端作为 push 端,而 client 端作为 pull 端.如果有多个 client 端同时连接到这个 server,则 服务器会在内部做一个负载均衡,采用平均分配的算法,将所有的消息均衡发布到 client 端上. 看上去,很稀松平常?接下来亮点真的来了
9、. 2),3)两种模式中,无论是 server 端还是 client 端在启动时启动时 都是不知道 client 实际数量的,这就 意味着,一个使用 zeromq 搭建的服务,可以进行”热更新”. 考虑如下一种场景.一个 server 端做为一组服务器集群最上层的一个 proxy,起到负载均衡的 作用,将请求按照它下面对应服务器集群依次派发到不同的 client 端进行处理.某个时刻可能 处理的机器只有2台,而随着负载越来越大,可能需要3台机器了,这个时候如果使用 zeromq 的 push-pull 搭建的 proxy 端,则可以不用对之前搭建的 server,client 端进行停机,只需
10、要新启动 一个 client 连接上去,proxy 层就会自动根据当 前的机器分配平均派发任务了.cool. 实际上,这些模式并不是什么新东西,只不过 zeromq 为使用者做了一个封装,而不是像 libevent,ACE 等还局限在网络层部分的封装,它关注的是通信层,通信模式等.4搭建安装 Linux 安装 下载地址：http:/download.zeromq.org/zeromq-2.1.7.tar.gz 解压后下载文件 cd zeromq-2.1.7 ./configure -prefix=/usr/local/zeromq make make installWindow 安装： 下载地
11、址：http:/download.zeromq.org/zeromq-2.1.7.zip 需要 C 编译环境，VC2008 或者 MinGW，RubyInstall 中的 DevKit 下载 MinGW 地址： http:/ inst/mingw-get-inst-20110530/mingw-get-inst-20110530.exe/download 点击打开，根据需要选择下载的功能，开始安装下载（需要网络） 安装完毕后，在 window 环境变量中设置 PATH 路径 下载 DevKit 地址：http:/rubyinstaller.org/downloads 下载 exe 文件安装即可
12、，在 Window 环境变量中设置 PATH 路径5支持的客户端C，C+，JAVA，PHP，Python，Ruby，C#，Erlang，Perl 等等6运行(Linux 系统，Java 为例)前提：需要定义环境变量 JAVA_HOME需要先安装 zeromq 下载 jzero：https:/ 解压下载文件 cd zeromq-q-59616d9 $ ./autogen.sh $ ./configure -prefix=/opt/q -with-zeromq=/usr/local/zeromq 回车运行后，如果缺少什么会有提示，根据提示解决就可以 比如 zmq.h 缺少，那就是证明 -with-
13、zeromq 配置不正确 Java 找不到，说 JAVA_HOME 没有设置正确 $ make $ make install编译完成，编写测试 java 文件 package com.q; import org.zeromq.ZMQ; import org.zeromq.ZMQQueue; public class MqQueue public static void main(String args) ZMQ.Context context = ZMQ.context(1); ZMQ.Socket clients = context.socket(ZMQ.XREP); clients.bin
14、d(“inproc:/gate_clients“); ZMQ.Socket workers = context.socket(ZMQ.XREQ); workers.bind(“inproc:/gate_workers“); ZMQ.Socket client = context.socket(ZMQ.REQ); client.connect(“inproc:/gate_clients“); ZMQ.Socket worker = context.socket(ZMQ.REP); worker.connect(“inproc:/gate_workers“); Thread t = new Thr
15、ead(new ZMQQueue(context, clients, workers); t.start(); for (int i = 0; i 100; i+) byte req = (“request“ + i).getBytes(); byte rsp = (“response“ + i).getBytes(); client.send(req, 0); / worker receives request byte reqTmp = worker.recv(0); System.out.println(new String(reqTmp); / worker sends respons
16、e worker.send(rsp, 0); / client receives response byte rspTmp = client.recv(0); System.out.println(new String(rspTmp); t.interrupt(); 编译前准备 设置 export JAVA_HOME=/usr/java/jdk1.6.0.22 设置 export CLASSPATH=$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar:/usr/loc al/zeromq/share/java/zmq.jar:.设置 export LD_LIBR
