ZMQ是啥？咋用？有教程吗？有demo吗？|||有|||
Posted 我要出家当道士
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ZMQ是啥？咋用？有教程吗？有demo吗？|||有|||相关的知识，希望对你有一定的参考价值。
目录
一、基本介绍
ZMQ 是一个异步的可嵌入的网络库，支持进程内、进程间、TCP 和 多播的通信。基于socket支持fan-out，pub-sub，分布式任务和req-res模式。异步的消息处理模型。官网的介绍非常详细，下面贴出官网链接和开发文档链：
ZeroMQ；Introduction | ØMQ - The Guide；ZeroMQ API - 0MQ Api
我在项目中主要使用了其中的发布订阅功能，实现进程间的通讯。基于domain socket的发布订阅功能，通过简单的测试，在保障不丢包的情况下其吞吐量大概在5Gb/s左右，发布订阅延迟在100微秒左右，能够满足项目中进程间的通讯需求。测试只是粗略的评估，只作为参考。
二、安装
可在线安装，也可以源码编译，不依赖第三方源，打开终端直接下载即可。
sudo apt-get install libzmq3-dev三、Demo
0、基本介绍与gcc编译
我是使用 C 进行开发的，有意思的是官网提供了ZMQ的C版本CZMQ，当然libzmq也支持C。具体看链接：ZeroMQ | C。两者的区别在于CZMQ提供了更高级的API，libzmq比较底层。我个人推荐使用 libzmq，CZMQ更新比较慢，使用的人很少，出现问题也不好解决。
下文中将提供两种通讯模型的demo，当然去官网也可以看到。代码中依赖的头文件内容贴在文末了。
编写好C代码之后，可以直接使用gcc编译。
gcc demo.c -o demo -l zmq1、request与response
这个通讯模式应该是最常用的了吧，大家应该都了解。客户端发出请求，服务端进行应答。
client.c
// Hello World client
#include <zmq.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
int main (void)
printf ("Connecting to hello world server...\\n");
void *context = zmq_ctx_new ();
void *requester = zmq_socket (context, ZMQ_REQ);
zmq_connect (requester, "tcp://localhost:5555");
int request_nbr;
for (request_nbr = 0; request_nbr != 10; request_nbr++)
char buffer [10];
printf ("Sending Hello %d...\\n", request_nbr);
zmq_send (requester, "Hello", 5, 0);
zmq_recv (requester, buffer, 10, 0);
printf ("Received World %d\\n", request_nbr);
zmq_close (requester);
zmq_ctx_destroy (context);
return 0;
server.c
// Binds REP socket to tcp://*:5555
// Expects "Hello" from client, replies with "World"
//
#include <zmq.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <assert.h>
int main (void)
// Socket to talk to clients
void *context = zmq_ctx_new ();
void *responder = zmq_socket (context, ZMQ_REP);
int rc = zmq_bind (responder, "tcp://*:5555");
assert (rc == 0);
while (1)
char buffer [10];
zmq_recv (responder, buffer, 10, 0);
printf ("Received Hello\\n");
sleep (1); // Do some 'work'
zmq_send (responder, "World", 5, 0);
return 0;
按官网说，该模型可同时支持数千个client连接server。有意思的是，当我先启动client，后启动server，两个程序也都是正常收发，猜测client有不断的重连机制和缓冲区。正常收发过程中，kill掉server，client会阻塞，应该是阻塞在recv了，可以通过设置非阻塞的recv来避免。
2、publish与subscribe
发布订阅通讯模型，订阅者与发布者建立连接后，发布者有需要传达的消息可以直接发布出去，所有订阅了该发布者的订阅者都可以收到该消息。具体看下面框架图。
publisher.c
// Weather update server
// Binds PUB socket to tcp://*:5556
// Publishes random weather updates
#include "zhelpers.h"
int main (void)
// Prepare our context and publisher
void *context = zmq_ctx_new ();
void *publisher = zmq_socket (context, ZMQ_PUB);
int rc = zmq_bind (publisher, "ipc:///tmp/feeds");
assert (rc == 0);
while (1)
// Send message to all subscribers
char update [20] = "hello";
s_send (publisher, update);
zmq_close (publisher);
zmq_ctx_destroy (context);
return 0;
subscriber.c
// Weather update client
// Connects SUB socket to tcp://localhost:5556
// Collects weather updates and finds avg temp in zipcode
#include "zhelpers.h"
int main (int argc, char *argv [])
// Socket to talk to server
printf ("Collecting updates from weather server...\\n");
void *context = zmq_ctx_new ();
void *subscriber = zmq_socket (context, ZMQ_SUB);
int rc = zmq_connect (subscriber, "ipc:///tmp/feeds");
assert (rc == 0);
// 必须调用，否则屏蔽所有消息；
// 不为空时，过滤满足该前缀的消息；
// 为空时，接收所有消息。
rc = zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE, NULL, 0);
assert (rc == 0);
while(1)
char *string = s_recv (subscriber);
printf("subscriibe : %s\\n", string);
free (string);
zmq_close (subscriber);
zmq_ctx_destroy (context);
return 0;
zhelpers.h
/* =====================================================================
zhelpers.h
Helper header file for example applications.
=====================================================================
*/
#ifndef __ZHELPERS_H_INCLUDED__
#define __ZHELPERS_H_INCLUDED__
// Include a bunch of headers that we will need in the examples
#include <zmq.h>
#include <assert.h>
#include <signal.h>
#include <stdarg.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#if (!defined (WIN32))
# include <sys/time.h>
#endif
#if (defined (WIN32))
# include <windows.h>
#endif
// Version checking, and patch up missing constants to match 2.1
#if ZMQ_VERSION_MAJOR == 2
# error "Please upgrade to ZeroMQ/3.2 for these examples"
#endif
// On some version of Windows, POSIX subsystem is not installed by default.
// So define srandom and random ourself.
#if (defined (WIN32))
# define srandom srand
# define random rand
#endif
// Provide random number from 0..(num-1)
#define randof(num) (int) ((float) (num) * random () / (RAND_MAX + 1.0))
// Receive 0MQ string from socket and convert into C string
// Caller must free returned string. Returns NULL if the context
// is being terminated.
static char *
s_recv (void *socket)
enum cap = 256 ;
char buffer [cap];
int size = zmq_recv (socket, buffer, cap - 1, 0);
if (size == -1)
return NULL;
buffer[size < cap ? size : cap - 1] = '\\0';
#if (defined (WIN32))
return strdup (buffer);
#else
return strndup (buffer, sizeof(buffer) - 1);
#endif
// remember that the strdup family of functions use malloc/alloc for space for the new string. It must be manually
// freed when you are done with it. Failure to do so will allow a heap attack.
// Convert C string to 0MQ string and send to socket
static int
s_send (void *socket, char *string)
int size = zmq_send (socket, string, strlen (string), 0);
return size;
// Sends string as 0MQ string, as multipart non-terminal
static int
s_sendmore (void *socket, char *string)
int size = zmq_send (socket, string, strlen (string), ZMQ_SNDMORE);
return size;
// Receives all message parts from socket, prints neatly
//
static void
s_dump (void *socket)
int rc;
zmq_msg_t message;
rc = zmq_msg_init (&message);
assert (rc == 0);
puts ("----------------------------------------");
// Process all parts of the message
do
int size = zmq_msg_recv (&message, socket, 0);
assert (size >= 0);
// Dump the message as text or binary
char *data = (char*)zmq_msg_data (&message);
assert (data != 0);
int is_text = 1;
int char_nbr;
for (char_nbr = 0; char_nbr < size; char_nbr++)
if ((unsigned char) data [char_nbr] < 32
|| (unsigned char) data [char_nbr] > 126)
is_text = 0;
printf ("[%03d] ", size);
for (char_nbr = 0; char_nbr < size; char_nbr++)
if (is_text)
printf ("%c", data [char_nbr]);
else
printf ("%02X", (unsigned char) data [char_nbr]);
printf ("\\n");
while (zmq_msg_more (&message));
rc = zmq_msg_close (&message);
assert (rc == 0);
#if (!defined (WIN32))
// Set simple random printable identity on socket
// Caution:
// DO NOT call this version of s_set_id from multiple threads on MS Windows
// since s_set_id will call rand() on MS Windows. rand(), however, is not
// reentrant or thread-safe. See issue #521.
static void
s_set_id (void *socket)
char identity [10];
sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));
zmq_setsockopt (socket, ZMQ_IDENTITY, identity, strlen (identity));
#else
// Fix #521 for MS Windows.
static void
s_set_id(void *socket, intptr_t id)
char identity [10];
sprintf(identity, "%04X", (int)id);
zmq_setsockopt(socket, ZMQ_IDENTITY, identity, strlen(identity));
#endif
// Sleep for a number of milliseconds
static void
s_sleep (int msecs)
#if (defined (WIN32))
Sleep (msecs);
#else
struct timespec t;
t.tv_sec = msecs / 1000;
t.tv_nsec = (msecs % 1000) * 1000000;
nanosleep (&t, NULL);
#endif
// Return current system clock as milliseconds
static int64_t
s_clock (void)
#if (defined (WIN32))
SYSTEMTIME st;
GetSystemTime (&st);
return (int64_t) st.wSecond * 1000 + st.wMilliseconds;
#else
struct timeval tv;
gettimeofday (&tv, NULL);
return (int64_t) (tv.tv_sec * 1000 + tv.tv_usec / 1000);
#endif
// Print formatted string to stdout, prefixed by date/time and
// terminated with a newline.
static void
s_console (const char *format, ...)
time_t curtime = time (NULL);
struct tm *loctime = localtime (&curtime);
char *formatted = (char*)malloc (20);
strftime (formatted, 20, "%y-%m-%d %H:%M:%S ", loctime);
printf ("%s", formatted);
free (formatted);
va_list argptr;
va_start (argptr, format);
vprintf (format, argptr);
va_end (argptr);
printf ("\\n");
#endif // __ZHELPERS_H_INCLUDED__以上是关于ZMQ是啥？咋用？有教程吗？有demo吗？|||有|||的主要内容，如果未能解决你的问题，请参考以下文章
ZMQ 套接字不是线程安全的，但我可以在不同的线程中使用 zmq_send() 和 zmq_recv() 吗？