ZMQ是什么？
#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 << "Connecting to hello world server..." << std::endl; //连接服务器 socket.connect ("tcp://localhost:5555"); for (int request_nbr = 0; request_nbr != 10; request_nbr++) { s_send (socket, "hello"); std::cout << "Client1 Received :" <<s_recv (socket)<< std::endl; Sleep(1000); } return 0; } DWORD WINAPI MyThread_client1(LPVOID lpParamter) { zmq::context_t context (1); //建立套接字 zmq::socket_t socket (context, ZMQ_REQ); std::cout << "Connecting to hello world server..." << std::endl; //连接服务器 socket.connect ("tcp://localhost:5555"); for (int request_nbr = 0; request_nbr != 10; request_nbr++) { s_send (socket, "SB"); std::cout << "Client2 Received :" <<s_recv (socket)<< std::endl; Sleep(1000); } return 0; } DWORD WINAPI MyThread_servce(LPVOID lpParamter) { zmq::context_t context (1); zmq::socket_t socket (context, ZMQ_REP); //绑定端口 socket.bind ("tcp://*:5555"); while (true) { std::cout << "Servce Received: "<<s_recv (socket)<< std::endl; s_send (socket, "world"); } } int main () { HANDLE hThread1 = CreateThread(NULL, 0, MyThread_client, NULL, 0, NULL); HANDLE hThread2 = CreateThread(NULL, 0, MyThread_servce, NULL, 0, NULL); HANDLE hThread3 = CreateThread(NULL, 0, MyThread_client1, NULL, 0, NULL); while(1); return 0; }
这里我建立了两个客户端和一个服务器，每个都独立运行一个线程。客户端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) { // Read envelope with address std::string address = s_recv (subscriber); // Read message contents std::string contents = s_recv (subscriber); std::cout << "订阅1：[" << address << "] " << contents << std::endl; } return 0; } //订阅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) { // Read envelope with address std::string address = s_recv (subscriber); // Read message contents std::string contents = s_recv (subscriber); std::cout << "订阅2：[" << address << "] " << contents << std::endl; } return 0; } //订阅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) { // Read envelope with address std::string address = s_recv (subscriber); // Read message contents std::string contents = s_recv (subscriber); std::cout << "订阅3：[" << address << "] " << contents << std::endl; } return 0; } //发布线程 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"); while (1) { // Write two messages, each with an envelope and content s_sendmore (publisher, "A"); s_send (publisher, "We don't want to see this"); Sleep (100); s_sendmore (publisher, "B"); s_send (publisher, "We would like to see this"); Sleep (100); } } 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; }
ZMQ API
ZMQ提供的所有API均以zmq_开头，
#include <zmq.h> gcc [flags] files -lzmq [libraries]
例如，返回当前ZMQ库的版本信息
void zmq_version (int *major, int *minor, int *patch);
Context
在使用任何ZQM库函数之前，必须首先创建ZMQ context（上下文），程序终止时，也需要销毁context。
创建context
void *zmq_ctx_new ();
ZMQ context是线程安全的，可以在多线程环境使用，而不需要程序员对其加/解锁。
在一个进程中，可以有多个ZMQ context并存。
设置context选项
int zmq_ctx_set (void *context, int option_name, int option_value); int zmq_ctx_get (void *context, int option_name);
销毁context
int zmq_ctx_term (void *context);
Sockets
ZMQ Sockets 是代表异步消息队列的一个抽象，注意，这里的ZMQ socket和POSIX套接字的socket不是一回事，ZMQ封装了物理连接的底层细节，对用户不透明。
传统的POSIX套接字只能支持1对1的连接，而ZMQ socket支持多个Client的并发连接，甚至在没有任何对端（peer）的情况下，ZMQ sockets上也能放入消息；
ZMQ sockets不是线程安全的，因此，不要在多个线程中并行操作同一个sockets。
创建ZMQ Sockets
void *zmq_socket (void *context, int type);
注意，ZMQ socket在bind之前还不能使用。
|
pattern |
type |
description |
|
一对一结对模型 |
ZMQ_PAIR |
|
|
请求回应模型 |
ZMQ_REQ |
client端使用 |
|
ZMQ_REP |
server端使用 |
|
|
ZMQ_DEALER |
将消息以轮询的方式分发给所有对端（peers） |
|
|
ZMQ_ROUTER |
||
|
发布订阅模型 |
ZMQ_PUB |
publisher端使用 |
|
ZMQ_XPUB |
||
|
ZMQ_SUB |
subscriber端使用 |
|
|
ZMQ_XSUB |
||
|
管道模型 |
ZMQ_PUSH |
push端使用 |
|
ZMQ_PULL |
pull端使用 |
|
|
原生模型 |
ZMQ_STREAM |
设置socket选项
int zmq_getsockopt (void *socket, int option_name, void *option_value, size_t *option_len); int zmq_setsockopt (void *socket, int option_name, const void *option_value, size_t option_len);
关闭socket
int zmq_close (void *socket);
创建一个消息流
int zmq_bind (void *socket, const char *endpoint); int zmq_connect (void *socket, const char *endpoint);
bind函数是将socket绑定到本地的端点（endpoint），而connect函数连接到指定的peer端点。
endpoint支持的类型：
|
transports |
description |
uri example |
|
zmp_tcp |
TCP的单播通信 |
tcp://*:8080 |
|
zmp_ipc |
本地进程间通信 |
ipc:// |
|
zmp_inproc |
本地线程间通信 |
inproc:// |
|
zmp_pgm |
PGM广播通信 |
pgm:// |
收发消息
int zmq_send (void *socket, void *buf, size_t len, int flags); int zmq_recv (void *socket, void *buf, size_t len, int flags); int zmq_send_const (void *socket, void *buf, size_t len, int flags);
zmq_recv()函数的len参数指定接收buf的最大长度，超出部分会被截断，函数返回的值是接收到的字节数，返回-1表示出错；
zmq_send()函数将指定buf的指定长度len的字节写入队列，函数返回值是发送的字节数，返回-1表示出错；
zmq_send_const()函数表示发送的buf是一个常量内存区（constant-memory），这块内存不需要复制、释放。
socket事件监控
int zmq_socket_monitor (void *socket, char * *addr, int events);
zmq_socket_monitor()函数会生成一对sockets，publishers端通过inproc://协议发布 sockets状态改变的events；
消息包含2帧，第1帧包含events id和关联值，第2帧表示受影响的endpoint。
监控支持的events：
ZMQ_EVENT_CONNECTED: 建立连接
ZMQ_EVENT_CONNECT_DELAYED: 连接失败
ZMQ_EVENT_CONNECT_RETRIED: 异步连接/重连
ZMQ_EVENT_LISTENING: bind到端点
ZMQ_EVENT_BIND_FAILED: bind失败
ZMQ_EVENT_ACCEPTED: 接收请求
ZMQ_EVENT_ACCEPT_FAILED: 接收请求失败
ZMQ_EVENT_CLOSED: 关闭连接
ZMQ_EVENT_CLOSE_FAILED: 关闭连接失败
ZMQ_EVENT_DISCONNECTED: 会话（tcp/ipc）中断
I/O多路复用
int zmq_poll (zmq_pollitem_t *items, int nitems, long timeout);
对sockets集合的I/O多路复用，使用水平触发。
与epoll类似，items参数指定一个结构体数组（结构体定义如下），nitems指定数组的元素个数，timeout参数是超时时间（单位：ms，0表示不等待立即返回，-1表示阻塞等待）。
typedef struct
{
void *socket;
int fd;
short events;
short revents;
} zmq_pollitem_t;
对于每个zmq_pollitem_t元素，ZMQ会同时检查其socket（ZMQ套接字）和fd（原生套接字）上是否有指定的events发生，且ZMQ套接字优先。
events指定该sockets需要关注的事件，revents返回该sockets已发生的事件，它们的取值为：
- ZMQ_POLLIN，可读；
- ZMQ_POLLOUT，可写；
- ZMQ_POLLERR，出错；
Messages
一个ZMQ消息就是一个用于在消息队列（进程内部或跨进程）中进行传输的数据单元，ZMQ消息本身没有数据结构，因此支持任意类型的数据，这完全依赖于程序员如何定义消息的数据结构。
一条ZMQ消息可以包含多个消息片（multi-part messages），每个消息片都是一个独立zmq_msg_t结构。
ZMQ保证以原子方式传递消息，要么所有消息片都发送成功，要么都不成功。
初始化消息
typedef void (zmq_free_fn) (void *data, void *hint); int zmq_msg_init (zmq_msg_t *msg); int zmq_msg_init_data (zmq_msg_t *msg, void *data, size_t size, zmq_free_fn *ffn, void *hint); int zmq_msg_init_size (zmq_msg_t *msg, size_t size);
zmq_msg_init()函数初始化一个消息对象zmq_msg_t ，不要直接访问zmq_msg_t对象，可以通过zmq_msg_* 函数来访问它。
zmq_msg_init()、zmq_msg_init_data()、zmq_msg_init_size() 三个函数是互斥的，每次使用其中一个即可。
设置消息属性
int zmq_msg_get (zmq_msg_t *message, int property); int zmq_msg_set (zmq_msg_t *message, int property, int value);
释放消息
int zmq_msg_close (zmq_msg_t *msg);
收发消息
int zmq_msg_send (zmq_msg_t *msg, void *socket, int flags); int zmq_msg_recv (zmq_msg_t *msg, void *socket, int flags);
其中，flags参数如下：
ZMQ_DONTWAIT，非阻塞模式，如果没有可用的消息，将errno设置为EAGAIN；
ZMQ_SNDMORE，发送multi-part messages时，除了最后一个消息片外，其它每个消息片都必须使用 ZMQ_SNDMORE 标记位。
获取消息内容
void *zmq_msg_data (zmq_msg_t *msg); int zmq_msg_more (zmq_msg_t *message); size_t zmq_msg_size (zmq_msg_t *msg);
zmq_msg_data()返回指向消息对象所带内容的指针；
zmq_msg_size()返回消息的字节数；
zmq_msg_more()标识该消息片是否是整个消息的一部分，是否还有更多的消息片待接收；
控制消息
int zmq_msg_copy (zmq_msg_t *dest, zmq_msg_t *src); int zmq_msg_move (zmq_msg_t *dest, zmq_msg_t *src);
zmq_msg_copy()函数实现的是浅拷贝；
zmq_msg_move()函数中，将dst指向src消息，然后src被置空。
eg，接收消息的代码示例：
zmq_msg_t part;
while (true) {
// Create an empty ØMQ message to hold the message part
int rc = zmq_msg_init (&part);
assert (rc == 0);
// Block until a message is available to be received from socket
rc = zmq_msg_recv (socket, &part, 0);
assert (rc != -1);
if (zmq_msg_more (&part))
fprintf (stderr, "more
");
else {
fprintf (stderr, "end
");
break;
}
zmq_msg_close (&part);
}
代理
ZMQ提供代理功能，代理可以在前端socket和后端socket之间转发消息。
int zmq_proxy (const void *frontend, const void *backend, const void *capture); int zmq_proxy_steerable (const void *frontend, const void *backend, const void *capture, const void *control);
共享队列（shared queue），前端是ZMQ_ROUTER socket，后端是ZMQ_DEALER socket，proxy会把clients发来的请求，公平地分发给services；
转发队列（forwarded），前端是ZMQ_XSUB socket, 后端是ZMQ_XPUB socket, proxy会把从publishers收到的消息转发给所有的subscribers；
流（streamer），前端是ZMQ_PULL socket, 后端是ZMQ_PUSH socket.
proxy使用的一个示例：
// Create frontend and backend sockets void *frontend = zmq_socket (context, ZMQ_ROUTER); assert (backend); void *backend = zmq_socket (context, ZMQ_DEALER); assert (frontend); // Bind both sockets to TCP ports assert (zmq_bind (frontend, "tcp://*:5555") == 0); assert (zmq_bind (backend, "tcp://*:5556") == 0); // Start the queue proxy, which runs until ETERM zmq_proxy frontend, backend, NULL);
错误处理
ZMQ库使用POSIX处理函数错误，返回NULL指针或者负数时表示调用出错。
int zmq_errno (void); const char *zmq_strerror (int errnum);
zmq_errno()函数返回当前线程的错误码errno变量的值；
zmq_strerror()函数将错误映射成错误字符串。
加密传输
ZQM可以为IPC和TCP连接提供安全机制：
- 不加密，zmq_null
- 使用用户名/密码授权，zmq_plain
- 椭圆加密，zmq_curve
这些通过 zmq_setsockopt()函数设置socket选项的时候配置。