消息队列之ZeroMQ(C++)

　　ZMQ是什么？

　　这是个类似于Socket的一系列接口，他跟Socket的区别是：普通的socket是端到端的（1:1的关系），而ZMQ却是可以N：M 的关系，人们对BSD套接字的了解较多的是点对点的连接，点对点连接需要显式地建立连接、销毁连接、选择协议（TCP/UDP）和处理错误等，而ZMQ屏蔽了这些细节，让你的网络编程更为简单。ZMQ用于node与node间的通信，node可以是主机或者是进程。

　　引用官方的说法： “ZMQ(以下ZeroMQ简称ZMQ)是一个简单好用的传输层，像框架一样的一个socket library，他使得Socket编程更加简单、简洁和性能更高。是一个消息处理队列库，可在多个线程、内核和主机盒之间弹性伸缩。ZMQ的明确目标是 “成为标准网络协议栈的一部分，之后进入Linux内核”。现在还未看到它们的成功。但是，它无疑是极具前景的、并且是人们更加需要的“传统”BSD套接字之上的一层封装。ZMQ让编写高性能网络应用程序极为简单和有趣。”

　　以上拷贝至百度百科。

　　对了使用ZMQ之前必须要有那么几样东西libzmq.lib，zhelpers.hpp，zmq.h，zmq.hpp。这些都可以在ZMQ的官网下载。

　　接下来是来说说ZMQ有模式，可以归纳成三种，请求回应模式（1对N），发布订阅模式(单向1对N)，还有推拉模型。

1：请求回应模式（Req-Rep）

可以有多个client，这个很容易理解跟TCP很像，但服务器与客户端必须是1问1答的形式。直接看源代码。

#include <zmq.hpp>
#include <string>
#include <iostream>
#include <windows.h>
#include<zhelpers.hpp>
using namespace std;
DWORD WINAPI MyThread_client(LPVOID lpParamter)

{

zmq::context_t context (1);

//建立套接字

    zmq::socket_t socket (context, ZMQ_REQ);
std::cout </span>&lt;&lt; <span style="color: #800000;">"</span><span style="color: #800000;">Connecting to hello world server...</span><span style="color: #800000;">"</span> &lt;&lt;<span style="color: #000000;"> std::endl;
</span><span style="color: #008000;">//</span><span style="color: #008000;">连接服务器</span>
socket.connect (<span style="color: #800000;">"</span><span style="color: #800000;">tcp://localhost:5555</span><span style="color: #800000;">"</span><span style="color: #000000;">);

</span><span style="color: #0000ff;">for</span> (<span style="color: #0000ff;">int</span> request_nbr = <span style="color: #800080;">0</span>; request_nbr != <span style="color: #800080;">10</span>; request_nbr++<span style="color: #000000;">) {
    s_send (socket, </span><span style="color: #800000;">"</span><span style="color: #800000;">hello</span><span style="color: #800000;">"</span><span style="color: #000000;">);
    std::cout </span>&lt;&lt; <span style="color: #800000;">"</span><span style="color: #800000;">Client1 Received :</span><span style="color: #800000;">"</span> &lt;&lt;s_recv (socket)&lt;&lt;<span style="color: #000000;"> std::endl;

    Sleep(</span><span style="color: #800080;">1000</span><span style="color: #000000;">);
}
</span><span style="color: #0000ff;">return</span> <span style="color: #800080;">0</span><span style="color: #000000;">;

}
DWORD WINAPI MyThread_client1(LPVOID lpParamter)

{

zmq::context_t context (1);

//建立套接字

    zmq::socket_t socket (context, ZMQ_REQ);
std::cout </span>&lt;&lt; <span style="color: #800000;">"</span><span style="color: #800000;">Connecting to hello world server...</span><span style="color: #800000;">"</span> &lt;&lt;<span style="color: #000000;"> std::endl;
</span><span style="color: #008000;">//</span><span style="color: #008000;">连接服务器</span>
socket.connect (<span style="color: #800000;">"</span><span style="color: #800000;">tcp://localhost:5555</span><span style="color: #800000;">"</span><span style="color: #000000;">);

</span><span style="color: #0000ff;">for</span> (<span style="color: #0000ff;">int</span> request_nbr = <span style="color: #800080;">0</span>; request_nbr != <span style="color: #800080;">10</span>; request_nbr++<span style="color: #000000;">) {
    s_send (socket, </span><span style="color: #800000;">"</span><span style="color: #800000;">SB</span><span style="color: #800000;">"</span><span style="color: #000000;">);
    std::cout </span>&lt;&lt; <span style="color: #800000;">"</span><span style="color: #800000;">Client2 Received :</span><span style="color: #800000;">"</span> &lt;&lt;s_recv (socket)&lt;&lt;<span style="color: #000000;"> std::endl;

    Sleep(</span><span style="color: #800080;">1000</span><span style="color: #000000;">);
}
</span><span style="color: #0000ff;">return</span> <span style="color: #800080;">0</span><span style="color: #000000;">;

}
DWORD WINAPI MyThread_servce(LPVOID lpParamter)

{

zmq::context_t context (1);

zmq::socket_t socket (context, ZMQ_REP);

//绑定端口

socket.bind ("tcp://*:5555");
</span><span style="color: #0000ff;">while</span> (<span style="color: #0000ff;">true</span><span style="color: #000000;">) {
    std::cout </span>&lt;&lt; <span style="color: #800000;">"</span><span style="color: #800000;">Servce Received: </span><span style="color: #800000;">"</span>&lt;&lt;s_recv (socket)&lt;&lt;<span style="color: #000000;"> std::endl;
    s_send (socket, </span><span style="color: #800000;">"</span><span style="color: #800000;">world</span><span style="color: #800000;">"</span><span style="color: #000000;">);
}

}
int main ()

{
HANDLE hThread1 </span>= CreateThread(NULL, <span style="color: #800080;">0</span>, MyThread_client, NULL, <span style="color: #800080;">0</span><span style="color: #000000;">, NULL);
HANDLE hThread2 </span>= CreateThread(NULL, <span style="color: #800080;">0</span>, MyThread_servce, NULL, <span style="color: #800080;">0</span><span style="color: #000000;">, NULL);

HANDLE hThread3 </span>= CreateThread(NULL, <span style="color: #800080;">0</span>, MyThread_client1, NULL, <span style="color: #800080;">0</span><span style="color: #000000;">, NULL);

</span><span style="color: #0000ff;">while</span>(<span style="color: #800080;">1</span><span style="color: #000000;">);
</span><span style="color: #0000ff;">return</span> <span style="color: #800080;">0</span><span style="color: #000000;">;

}

运行结果：

　　这里我建立了两个客户端和一个服务器，每个都独立运行一个线程。客户端1发了“hello”，客户端2发了“SB”，服务器都能接收到并且返回了world。

2：发布订阅模式（PUB-SUB)

所谓发布订阅，比如天气预报，当很多人订阅之后，中心服务器直接往订阅的人发送就可以了，不需要管对方有没有收到。也就是1对N的模式。这里还有重要的一个概念，频道：跟收音机的频道类似，订阅者设定了频道才能听到该频道的消息

看例程序：

#include <zmq.hpp>
#include <string>
#include <iostream>
#include <windows.h>
#include<zhelpers.hpp>
using namespace std;
//订阅1

DWORD WINAPI MyThread_sub1(LPVOID lpParamter)

{

zmq::context_t context(1);

zmq::socket_t subscriber (context, ZMQ_SUB);

//连接

subscriber.connect("tcp://localhost:5563");

//设置频道B

subscriber.setsockopt( ZMQ_SUBSCRIBE, "A", 1);

while (1) {
    </span><span style="color: #008000;">//</span><span style="color: #008000;">  Read envelope with address</span>
    std::<span style="color: #0000ff;">string</span> address =<span style="color: #000000;"> s_recv (subscriber);
    </span><span style="color: #008000;">//</span><span style="color: #008000;">  Read message contents</span>
    std::<span style="color: #0000ff;">string</span> contents =<span style="color: #000000;"> s_recv (subscriber);
    
    std::cout </span>&lt;&lt; <span style="color: #800000;">"</span><span style="color: #800000;">订阅1：[</span><span style="color: #800000;">"</span> &lt;&lt; address &lt;&lt; <span style="color: #800000;">"</span><span style="color: #800000;">] </span><span style="color: #800000;">"</span> &lt;&lt; contents &lt;&lt;<span style="color: #000000;"> std::endl;
}
</span><span style="color: #0000ff;">return</span> <span style="color: #800080;">0</span><span style="color: #000000;">;

}

//订阅2

DWORD WINAPI MyThread_sub2(LPVOID lpParamter)

{

zmq::context_t context(1);

zmq::socket_t subscriber (context, ZMQ_SUB);

//连接

subscriber.connect("tcp://localhost:5563");

//设置频道B

subscriber.setsockopt( ZMQ_SUBSCRIBE, "B", 1);

while (1) {
    </span><span style="color: #008000;">//</span><span style="color: #008000;">  Read envelope with address</span>
    std::<span style="color: #0000ff;">string</span> address =<span style="color: #000000;"> s_recv (subscriber);
    </span><span style="color: #008000;">//</span><span style="color: #008000;">  Read message contents</span>
    std::<span style="color: #0000ff;">string</span> contents =<span style="color: #000000;"> s_recv (subscriber);
    
    std::cout </span>&lt;&lt; <span style="color: #800000;">"</span><span style="color: #800000;">订阅2：[</span><span style="color: #800000;">"</span> &lt;&lt; address &lt;&lt; <span style="color: #800000;">"</span><span style="color: #800000;">] </span><span style="color: #800000;">"</span> &lt;&lt; contents &lt;&lt;<span style="color: #000000;"> std::endl;
}
</span><span style="color: #0000ff;">return</span> <span style="color: #800080;">0</span><span style="color: #000000;">;

}

//订阅3

DWORD WINAPI MyThread_sub3(LPVOID lpParamter)

{

zmq::context_t context(1);

zmq::socket_t subscriber (context, ZMQ_SUB);

//连接

subscriber.connect("tcp://localhost:5563");

//设置频道B

// subscriber.setsockopt( ZMQ_SUBSCRIBE, "B", 1);

while (1) {
    </span><span style="color: #008000;">//</span><span style="color: #008000;">  Read envelope with address</span>
    std::<span style="color: #0000ff;">string</span> address =<span style="color: #000000;"> s_recv (subscriber);
    </span><span style="color: #008000;">//</span><span style="color: #008000;">  Read message contents</span>
    std::<span style="color: #0000ff;">string</span> contents =<span style="color: #000000;"> s_recv (subscriber);
    
    std::cout </span>&lt;&lt; <span style="color: #800000;">"</span><span style="color: #800000;">订阅3：[</span><span style="color: #800000;">"</span> &lt;&lt; address &lt;&lt; <span style="color: #800000;">"</span><span style="color: #800000;">] </span><span style="color: #800000;">"</span> &lt;&lt; contents &lt;&lt;<span style="color: #000000;"> std::endl;
}
</span><span style="color: #0000ff;">return</span> <span style="color: #800080;">0</span><span style="color: #000000;">;

}

//发布线程

DWORD WINAPI MyThread_pub(LPVOID lpParamter)

{

//  Prepare our context and publisher

zmq::context_t context(1);

zmq::socket_t publisher(context, ZMQ_PUB);

publisher.bind("tcp://*:5563");
</span><span style="color: #0000ff;">while</span> (<span style="color: #800080;">1</span><span style="color: #000000;">) {
    </span><span style="color: #008000;">//</span><span style="color: #008000;">  Write two messages, each with an envelope and content</span>
    s_sendmore (publisher, <span style="color: #800000;">"</span><span style="color: #800000;">A</span><span style="color: #800000;">"</span><span style="color: #000000;">);
    s_send (publisher, </span><span style="color: #800000;">"</span><span style="color: #800000;">We don't want to see this</span><span style="color: #800000;">"</span><span style="color: #000000;">);

    Sleep (</span><span style="color: #800080;">100</span><span style="color: #000000;">);
    s_sendmore (publisher, </span><span style="color: #800000;">"</span><span style="color: #800000;">B</span><span style="color: #800000;">"</span><span style="color: #000000;">);
    s_send (publisher, </span><span style="color: #800000;">"</span><span style="color: #800000;">We would like to see this</span><span style="color: #800000;">"</span><span style="color: #000000;">);
    Sleep (</span><span style="color: #800080;">100</span><span style="color: #000000;">);

}

}
int main ()

{

HANDLE hThread1 = CreateThread(NULL, 0, MyThread_pub, NULL, 0, NULL);

Sleep(1000);

HANDLE hThread2 = CreateThread(NULL, 0, MyThread_sub1, NULL, 0, NULL);

HANDLE hThread3 = CreateThread(NULL, 0, MyThread_sub2, NULL, 0, NULL);

HANDLE hThread4 = CreateThread(NULL, 0, MyThread_sub3, NULL, 0, NULL);

while(1);

return 0;

}

结果：

例程中，设置了1个发布端和3个订阅端，订阅端订阅的频道分别是是A,B和没有订阅，发布端发布了对应频道的订阅消息。由此订阅3没有设置频道，所以收不到消息。

3：推拉模式，这个词语感觉污，还是我想歪了。还没研究过，不过看网上说的，跟发布订阅模式类似，只是可以负载均衡。目前项目中也没有用到，下次有机会再研究吧。

ZEROMQ官网的github上面有详细的例程：https://github.com/imatix/zguide/tree/master/examples/

无欲则刚关心则乱