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、提前分配该磁盘文件的空间，或者将每次分配的空间调大一些，这样可以避免磁盘碎片的产生，并保证读写是连续的。
另外，我不建议将消息保存在数据库中，甚至不建议交给那些所谓的高速键值缓存，它们比起一个磁盘文件要来得昂贵。
如果你想让巨人模式变得更为可靠，你可以将请求复制到另一台服务器上，这样就不需要担心主程序遭到核武器袭击了。
如果你想让巨人模式变得更为快速，但可以牺牲一些可靠性，那你可以将请求和应答都保存在内存中。这样做可以让该服务作为脱机网络运行，不过若巨人服务本身崩溃了，我也无能为力。
