ZMQ JAVA使用经验之 ZMQ简介怎么解决
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ZMQ JAVA使用经验之 ZMQ简介怎么解决相关的知识，希望对你有一定的参考价值。
ZMQ JAVA使用经验之 ZMQ简介怎么解决：ZMQ被称为史上最快消息队列，它处于会话层之上，应用层之下，使用后台异步线程完成消息的接受和发送，完美的封装了Socket API，大大简化了编程人员的复杂度，被称为史上最强大的消息中间件。ZMQ是用C语言编写的，30s内完成消息的传输，能够兼容多个平台，多种语言，可以使用多种方式实现N对N的Socket连接。本文仅以JAVA版本的ZMQ API为例，介绍ZMQ。
ZMQ与传统的TCP Socket相比，具有以下优点:
1） ZMQ发送和接受的是具有固定长度的二进制对象，ZMQ的消息包最大254个字节，前6个字节是协议，然后是数据包。数据包由3个部分组成，第一个字节是包的长度，第二个字节是包的一些属性，最后是包的内容。如果超过255个字节（有一个字节表示包属性），则ZMQ会自动分包传输；而对于TCP Socket，是面向字节流的连接，编程人员需要处理数据块与数据块之间的边界问题，而ZMQ能够保证每次用户发送和接受的都是一个完整的数据块；
2) 传统的TCP Socket的连接是1对1的，可以认为“1个Socket=1个连接”，每一个线程独立的维护一个Socket。但是ZMQ摒弃了这种1对1的模式，ZMQ的Socket可以很轻松的实现1对N，N对1和N对N的连接模式，一个ZMQ的Socket可以自动的维护一组连接，用户无法操作这些连接，用户只能操作套接字，而不是连接本身，所以说ZMQ的世界里，连接是私有的。这里大家关心的一点是，一个Socket是如何识别来自多个Socket的连接的，这里以请求响应模式为例介绍ZMQ是如何实现一个Socket连接N个Socket的；
3）ZMQ使用异步后台线程处理接受和发送请求，这意味着发送完消息，不可以立即释放资源，消息什么时候发送用户是无法控制的，同时，ZMQ自动重连，这意味着用户可以以任意顺序加入到网络中，服务器也可以随时加入或者退出网络；
ZMQ之所以能够在无状态的网络中实现1对N的连接，关键在于信封的机制，信封里保存了应答目标的位置。ZMQ涉及到请求-响应模式的Socket一共有4种类型：
DEALER是一种负载均衡，它会将消息分发给已连接的节点，并使用公平队列的机制处理接受到的消息。
REQ发送消息时会在消息顶部插入一个空帧，接受时会将空帧移去。其实REQ是建立在DEALER之上的，但REQ只有当消息发送并接受到回应后才能继续运行。
ROUTER在收到消息时会在顶部添加一个信封，标记消息来源。发送时会通过该信封决定哪个节点可以获取到该条消息。
REP在收到消息时会将第一个空帧之前的所有信息保存起来，将原始信息传送给应用程序。在发送消息时，REP会用刚才保存的信息包裹应答消息。REP其实是建立在ROUTER之上的，但和REQ一样，必须完成接受和发送这两个动作后才能继续。
在了解了4种类型的Socket之后，我们就不难理解ZMQ的信封机制了。ZMQ信封机制的核心是Router Socket，它的工作原理如下：
从ROUTER中读取一条消息时，ZMQ会包上一层信封，上面注明了消息的来源。向ROUTER写入一条消息时（包含信封），ZMQ会将信封拆开，并将消息递送给相应的对象。当REQ Socket向ROUTER Socket发送一条请求后，REP会从ROUTER收到一条消息，消息格式如下：
第三帧是REP从应用程序收到的数据，第二帧是空帧，是REQ在发送ROUTER数据之前添加的，用来表示结束，第一帧即信封，是ROUTER添加的，主要用来记录消息来源；整个数据包处理过程如下：
对于REQ Socket，可以在创建Socket的时候，为该Sock指定标示符，此时的Socket称为持久Socket，没有指定标示符的我们称为瞬时Socket，ROUTER会自动为瞬时Socket生成一个标示符；
这样REP返回包含信封的数据给ROUTER，ROUTER就可以根据信封上的标示符将该消息发送到对应的REQ上；
ZMQ使用注意事项：
ZMQ是在发送端缓存消息的，可以通过阈值控制消息的溢出；
ZMQ不可以线程之间共享Socket，否则会报org.zeromq.ZMQException: Operation cannot be accomplished in current state错误。
ZMQ一个进程只允许有一个Context，new Context(int arg) arg表示后台线程的数量；
ZMQ的Socket类有一个Linger参数，可以通过SetLinger设置，主要用于表示该Socket关闭以后，未发送成功的消息是否还保存，如果设置为-1表示该消息将永久保存（除非宕机，ZMQ是不持久化消息的），如果为0表示所有未发送成功的消息在Socker关闭以后都将立即清除，如果是一个正数，则表示该消息在Socket关闭后多少毫秒内被删除；这个方法非常有用，尤其在控制发送失败时，是否重发消息。 参考技术A 1） ZMQ发送和接受的是具有固定长度的二进制对象，ZMQ的消息包最大254个字节，前6个字节是协议，然后是数据包。数据包由3个部分组成，第一个字节是包的长度，第二个字节是包的一些属性，最后是包的内容。如果超过255个字节（有一个字节表示包属性），则ZMQ会自动分包传输；而对于TCP Socket，是面向字节流的连接，编程人员需要处理数据块与数据块之间的边界问题，而ZMQ能够保证每次用户发送和接受的都是一个完整的数据块；
2) 传统的TCP Socket的连接是1对1的，可以认为“1个Socket=1个连接”，每一个线程独立的维护一个Socket。但是ZMQ摒弃了这种1对1的模式，ZMQ的Socket可以很轻松的实现1对N，N对1和N对N的连接模式，一个ZMQ的Socket可以自动的维护一组连接，用户无法操作这些连接，用户只能操作套接字，而不是连接本身，所以说ZMQ的世界里，连接是私有的。这里大家关心的一点是，一个Socket是如何识别来自多个Socket的连接的，这里以请求响应模式为例介绍ZMQ是如何实现一个Socket连接N个Socket的；
3）ZMQ使用异步后台线程处理接受和发送请求，这意味着发送完消息，不可以立即释放资源，消息什么时候发送用户是无法控制的，同时，ZMQ自动重连，这意味着用户可以以任意顺序加入到网络中，服务器也可以随时加入或者退出网络；
ZMQ之脱机可靠性--巨人模式
当你意识到管家模式是一种非常可靠的消息代理时，你可能会想要使用磁盘做一下消息中转，从而进一步提升可靠性。这种方式虽然在很多企业级消息系统中应用，但我还是有些反对的，原因有：
1、我们可以看到，懒惰海盗模式的client可以工作得非常好，能够在多种架构中运行。唯一的问题是它会假设worker是无状态的，且提供的服务是幂等的。但这个问题我们可以通过其他方式解决，而不是添加磁盘。
2、添加磁盘会带来新的问题，需要额外的管理和维护费用。海盗模式的最大优点就是简单明了，不会崩溃。如果你还是担心硬件会出问题，可以改用点对点的通信模式，这会在本章最后一节讲到。
虽然有以上原因，但还是有一个合理的场景可以用到磁盘中转的——异步脱机网络。海盗模式有一个问题，那就是client发送请求后会一直等待应答。如果client和worker并不是长连接（可以拿电子邮箱做个类比），我们就无法在client和worker之间建立一个无状态的网络，因此需要将这种状态保存起来。
于是我们就有了巨人模式，该模式下会将消息写到磁盘中，确保不会丢失。当我们进行服务查询时，会转向巨人这一层进行。巨人是建立在管家之上的，而不是改写了MDP协议。这样做的好处是我们可以在一个特定的worker中实现这种可靠性，而不用去增加代理的逻辑。
实现更为简单：
1、代理用一种语言编写，worker使用另一种语言编写。
2、可以自由升级这种模式。
唯一的缺点是，代理和磁盘之间会有一层额外的联系，不过这也是值得的。
我们有很多方法来实现一种持久化的请求-应答架构，而目标当然是越简单越好。我能想到的最简单的方式是提供一种成为“巨人”的代理服务，它不会影响现有worker的工作，若client想要立即得到应答，它可以和代理进行通信；如果它不是那么着急，那就可以和巨人通信：“嗨，巨人，麻烦帮我处理下这个请求，我去买些菜。”
这样一来，巨人就既是worker又是client。client和巨人之间的对话一般是：
1、Client: 请帮我处理这个请求。巨人：好的。
2、Client: 有要给我的应答吗？巨人：有的。（或者没有）
3、Client: OK，你可以释放那个请求了，工作已经完成。巨人：好的。
巨人和代理之间的对话一般是：
1、巨人：嗨，代理程序，你这里有个叫echo的服务吗？代理：恩，好像有。
2、巨人：嗨，echo服务，请帮我处理一下这个请求。Echo: 好了，这是应答。
3、巨人：谢谢！
你可以想象一些发生故障的情形，看看上述模式是否能解决？worker在处理请求的时候崩溃，巨人会不断地重新发送请求；应答在传输过程中丢失了，巨人也会重试；如果请求已经处理，但client没有得到应答，那它会再次询问巨人；如果巨人在处理请求或进行应答的时候崩溃了，客户端会进行重试；只要请求是被保存在磁盘上的，那它就不会丢失。
这个机制中，握手的过程是比较漫长的，但client可以使用异步的管家模式，一次发送多个请求，并一起等待应答。
我们需要一种方法，让client会去请求应答内容。不同的client会访问到相同的服务，且client是来去自由的，有着不同的标识。一个简单、合理、安全的解决方案是：
1、当巨人收到请求时，它会为每个请求生成唯一的编号（UUID），并将这个编号返回给client；
2、client在请求应答内容时需要提供这个编号。
这样一来client就需要负责将UUID安全地保存起来，不过这就省去了验证的过程。有其他方案吗？我们可以使用持久化的套接字，即显式声明客户端的套接字标识。然而，这会造成管理上的麻烦，而且万一两个client的套接字标识相同，那会引来无穷的麻烦。
在我们开始制定一个新的协议之前，我们先思考一下client如何和巨人通信。一种方案是提供一种服务，配合三个不同的命令；另一种方案则更为简单，提供三种独立的服务：
1、titanic.request - 保存一个请求，并返回UUID
2、titanic.reply - 根据UUID获取应答内容
3、titanic.close - 确认某个请求已被正确地处理
我们需要创建一个多线程的worker，正如我们之前用ZMQ进行多线程编程一样，很简单。但是，在我们开始编写代码之前，先讲巨人模式的一些定义写下来：http://rfc.zeromq.org/spec:9 。我们称之为“巨人服务协议”，或TSP。
使用TSP协议自然会让client多出额外的工作，下面是一个简单但足够健壮的client：
ticlient: Titanic client example in C
//
// 巨人模式client示例
// 实现 http://rfc.zeromq.org/spec:9 协议中的client端
// 让我们直接编译，不创建类库
#include "mdcliapi.c"
// 请求TSP协议下的服务
// 如果成功则返回应答（状态码：200），否则返回NULL
//
static zmsg_t *
s_service_call (mdcli_t *session, char *service, zmsg_t **request_p)
zmsg_t *reply = mdcli_send (session, service, request_p);
if (reply)
zframe_t *status = zmsg_pop (reply);
if (zframe_streq (status, "200"))
zframe_destroy (&status);
return reply;
else
if (zframe_streq (status, "400"))
printf ("E: 客户端发生严重错误，取消请求\\n");
exit (EXIT_FAILURE);
else
if (zframe_streq (status, "500"))
printf ("E: 服务端发生严重错误，取消请求\\n");
exit (EXIT_FAILURE);
else
exit (EXIT_SUCCESS); // 中断或发生错误
zmsg_destroy (&reply);
return NULL; // 请求不成功，但不返回失败原因
int main (int argc, char *argv [])
int verbose = (argc > 1 && streq (argv [1], "-v"));
mdcli_t *session = mdcli_new ("tcp://localhost:5555", verbose);
// 1. 发送echo服务的请求给巨人
zmsg_t *request = zmsg_new ();
zmsg_addstr (request, "echo");
zmsg_addstr (request, "Hello world");
zmsg_t *reply = s_service_call (
session, "titanic.request", &request);
zframe_t *uuid = NULL;
if (reply)
uuid = zmsg_pop (reply);
zmsg_destroy (&reply);
zframe_print (uuid, "I: request UUID ");
// 2. 等待应答
while (!zctx_interrupted)
zclock_sleep (100);
request = zmsg_new ();
zmsg_add (request, zframe_dup (uuid));
zmsg_t *reply = s_service_call (
session, "titanic.reply", &request);
if (reply)
char *reply_string = zframe_strdup (zmsg_last (reply));
printf ("Reply: %s\\n", reply_string);
free (reply_string);
zmsg_destroy (&reply);
// 3. 关闭请求
request = zmsg_new ();
zmsg_add (request, zframe_dup (uuid));
reply = s_service_call (session, "titanic.close", &request);
zmsg_destroy (&reply);
break;
else
printf ("I: 尚未收到应答，准备稍后重试...\\n");
zclock_sleep (5000); // 5秒后重试
zframe_destroy (&uuid);
mdcli_destroy (&session);
return 0;
当然，上面的代码可以整合到一个框架中，程序员不需要了解其中的细节。如果我有时间的话，我会尝试写一个这样的API的，让应用程序又变回短短的几行。这种理念和MDP中的一致：不要做重复的事。
下面是巨人的实现。这个服务端会使用三个线程来处理三种服务。它使用最原始的持久化方法来保存请求：为每个请求创建一个磁盘文件。虽然简单，但也挺恐怖的。比较复杂的部分是，巨人会维护一个队列来保存这些请求，从而避免重复地扫描目录。
titanic: Titanic broker example in C
//
// 巨人模式 - 服务
//
// 实现 http://rfc.zeromq.org/spec:9 协议的服务端
// 让我们直接编译，不创建类库
#include "mdwrkapi.c"
#include "mdcliapi.c"
#include "zfile.h"
#include <uuid/uuid.h>
// 返回一个可打印的唯一编号（UUID）
// 调用者负责释放UUID字符串的内存
static char *
s_generate_uuid (void)
char hex_char [] = "0123456789ABCDEF";
char *uuidstr = zmalloc (sizeof (uuid_t) * 2 + 1);
uuid_t uuid;
uuid_generate (uuid);
int byte_nbr;
for (byte_nbr = 0; byte_nbr < sizeof (uuid_t); byte_nbr++)
uuidstr [byte_nbr * 2 + 0] = hex_char [uuid [byte_nbr] >> 4];
uuidstr [byte_nbr * 2 + 1] = hex_char [uuid [byte_nbr] & 15];
return uuidstr;
// 根据UUID生成用于保存请求内容的文件名，并返回
#define TITANIC_DIR ".titanic"
static char *
s_request_filename (char *uuid)
char *filename = malloc (256);
snprintf (filename, 256, TITANIC_DIR "/%s.req", uuid);
return filename;
// 根据UUID生成用于保存应答内容的文件名，并返回
static char *
s_reply_filename (char *uuid)
char *filename = malloc (256);
snprintf (filename, 256, TITANIC_DIR "/%s.rep", uuid);
return filename;
// ---------------------------------------------------------------------
// 巨人模式 - 请求服务
static void
titanic_request (void *args, zctx_t *ctx, void *pipe)
mdwrk_t *worker = mdwrk_new (
"tcp://localhost:5555", "titanic.request", 0);
zmsg_t *reply = NULL;
while (TRUE)
// 若应答非空则发送，再从代理处获得新的请求
zmsg_t *request = mdwrk_recv (worker, &reply);
if (!request)
break; // 中断并退出
// 确保消息目录是存在的
file_mkdir (TITANIC_DIR);
// 生成UUID，并将消息保存至磁盘
char *uuid = s_generate_uuid ();
char *filename = s_request_filename (uuid);
FILE *file = fopen (filename, "w");
assert (file);
zmsg_save (request, file);
fclose (file);
free (filename);
zmsg_destroy (&request);
// 将UUID加入队列
reply = zmsg_new ();
zmsg_addstr (reply, uuid);
zmsg_send (&reply, pipe);
// 将UUID返回给客户端
// 将由循环顶部的mdwrk_recv()函数完成
reply = zmsg_new ();
zmsg_addstr (reply, "200");
zmsg_addstr (reply, uuid);
free (uuid);
mdwrk_destroy (&worker);
// ---------------------------------------------------------------------
// 巨人模式 - 应答服务
static void *
titanic_reply (void *context)
mdwrk_t *worker = mdwrk_new (
"tcp://localhost:5555", "titanic.reply", 0);
zmsg_t *reply = NULL;
while (TRUE)
zmsg_t *request = mdwrk_recv (worker, &reply);
if (!request)
break; // 中断并退出
char *uuid = zmsg_popstr (request);
char *req_filename = s_request_filename (uuid);
char *rep_filename = s_reply_filename (uuid);
if (file_exists (rep_filename))
FILE *file = fopen (rep_filename, "r");
assert (file);
reply = zmsg_load (file);
zmsg_pushstr (reply, "200");
fclose (file);
else
reply = zmsg_new ();
if (file_exists (req_filename))
zmsg_pushstr (reply, "300"); //挂起
else
zmsg_pushstr (reply, "400"); //未知
zmsg_destroy (&request);
free (uuid);
free (req_filename);
free (rep_filename);
mdwrk_destroy (&worker);
return 0;
// ---------------------------------------------------------------------
// 巨人模式 - 关闭请求
static void *
titanic_close (void *context)
mdwrk_t *worker = mdwrk_new (
"tcp://localhost:5555", "titanic.close", 0);
zmsg_t *reply = NULL;
while (TRUE)
zmsg_t *request = mdwrk_recv (worker, &reply);
if (!request)
break; // 中断并退出
char *uuid = zmsg_popstr (request);
char *req_filename = s_request_filename (uuid);
char *rep_filename = s_reply_filename (uuid);
file_delete (req_filename);
file_delete (rep_filename);
free (uuid);
free (req_filename);
free (rep_filename);
zmsg_destroy (&request);
reply = zmsg_new ();
zmsg_addstr (reply, "200");
mdwrk_destroy (&worker);
return 0;
// 处理某个请求，成功则返回1
static int
s_service_success (mdcli_t *client, char *uuid)
// 读取请求内容，第一帧为服务名称
char *filename = s_request_filename (uuid);
FILE *file = fopen (filename, "r");
free (filename);
// 如果client已经关闭了该请求，则返回1
if (!file)
return 1;
zmsg_t *request = zmsg_load (file);
fclose (file);
zframe_t *service = zmsg_pop (request);
char *service_name = zframe_strdup (service);
// 使用MMI协议检查服务是否可用
zmsg_t *mmi_request = zmsg_new ();
zmsg_add (mmi_request, service);
zmsg_t *mmi_reply = mdcli_send (client, "mmi.service", &mmi_request);
int service_ok = (mmi_reply
&& zframe_streq (zmsg_first (mmi_reply), "200"));
zmsg_destroy (&mmi_reply);
if (service_ok)
zmsg_t *reply = mdcli_send (client, service_name, &request);
if (reply)
filename = s_reply_filename (uuid);
FILE *file = fopen (filename, "w");
assert (file);
zmsg_save (reply, file);
fclose (file);
free (filename);
return 1;
zmsg_destroy (&reply);
else
zmsg_destroy (&request);
free (service_name);
return 0;
int main (int argc, char *argv [])
int verbose = (argc > 1 && streq (argv [1], "-v"));
zctx_t *ctx = zctx_new ();
// 创建MDP客户端会话
mdcli_t *client = mdcli_new ("tcp://localhost:5555", verbose);
mdcli_set_timeout (client, 1000); // 1 秒
mdcli_set_retries (client, 1); // 只尝试一次
void *request_pipe = zthread_fork (ctx, titanic_request, NULL);
zthread_new (ctx, titanic_reply, NULL);
zthread_new (ctx, titanic_close, NULL);
// 主循环
while (TRUE)
// 如果没有活动，我们将每秒循环一次
zmq_pollitem_t items [] = request_pipe, 0, ZMQ_POLLIN, 0 ;
int rc = zmq_poll (items, 1, 1000 * ZMQ_POLL_MSEC);
if (rc == -1)
break; // 中断
if (items [0].revents & ZMQ_POLLIN)
// 确保消息目录是存在的
file_mkdir (TITANIC_DIR);
// 将UUID添加到队列中，使用“-”号标识等待中的请求
zmsg_t *msg = zmsg_recv (request_pipe);
if (!msg)
break; // 中断
FILE *file = fopen (TITANIC_DIR "/queue", "a");
char *uuid = zmsg_popstr (msg);
fprintf (file, "-%s\\n", uuid);
fclose (file);
free (uuid);
zmsg_destroy (&msg);
// 分派
//
char entry [] = "?.......:.......:.......:.......:";
FILE *file = fopen (TITANIC_DIR "/queue", "r+");
while (file && fread (entry, 33, 1, file) == 1)
// 处理UUID前缀为“-”的请求
if (entry [0] == -)
if (verbose)
printf ("I: 开始处理请求 %s\\n", entry + 1);
if (s_service_success (client, entry + 1))
// 标记为已处理
fseek (file, -33, SEEK_CUR);
fwrite ("+", 1, 1, file);
fseek (file, 32, SEEK_CUR);
// 跳过最后一行
if (fgetc (file) == \\r)
fgetc (file);
if (zctx_interrupted)
break;
if (file)
fclose (file);
mdcli_destroy (&client);
return 0;
测试时，打开mdbroker和titanic，再运行ticlient，然后开启任意个mdworker，就可以看到client获得了应答。
几点说明：
1、我们使用MMI协议去向代理询问某项服务是否可用，这一点和MDP中的逻辑一致；
2、我们使用inproc（进程内）协议建立主循环和titanic.request服务间的联系，保存新的请求信息。这样可以避免主循环不断扫描磁盘目录，读取所有请求文件，并按照时间日期排序。
这个示例程序不应关注它的性能（一定会非常糟糕，虽然我没有测试过），而是应该看到它是如何提供一种可靠的通信模式的。你可以测试一下，打开代理、巨人、worker和client，使用-v参数显示跟踪信息，然后随意地开关代理、巨人、或worker（client不能关闭），可以看到所有的请求都能获得应答。
如果你想在真实环境中使用巨人模式，你肯定会问怎样才能让速度快起来。以下是我的做法：
1、使用一个磁盘文件保存所有数据。操作系统处理大文件的效率要比处理许多小文件来的高。
2、使用一种循环的机制来组织该磁盘文件的结构，这样新的请求可以被连续地写入这个文件。单个线程在全速写入磁盘时的效率是比较高的。
3、将索引保存在内存中，可以在启动程序时重建这个索引。这样做可以节省磁盘缓存，让索引安全地保存在磁盘上。你需要用到fsync的机制来保存每一条数据；或者可以等待几毫秒，如果不怕丢失上千条数据的话。
4、如果条件允许，应选择使用固态硬盘；
5、提前分配该磁盘文件的空间，或者将每次分配的空间调大一些，这样可以避免磁盘碎片的产生，并保证读写是连续的。
另外，我不建议将消息保存在数据库中，甚至不建议交给那些所谓的高速键值缓存，它们比起一个磁盘文件要来得昂贵。
如果你想让巨人模式变得更为可靠，你可以将请求复制到另一台服务器上，这样就不需要担心主程序遭到核武器袭击了。
如果你想让巨人模式变得更为快速，但可以牺牲一些可靠性，那你可以将请求和应答都保存在内存中。这样做可以让该服务作为脱机网络运行，不过若巨人服务本身崩溃了，我也无能为力。
以上是关于ZMQ JAVA使用经验之 ZMQ简介怎么解决的主要内容，如果未能解决你的问题，请参考以下文章
如何解决“VisibleDeprecationWarning：zmq.eventloop.minitornado 已弃用”？