可靠的请求-应答模式

第三章中我们使用实例介绍了高级请求-应答模式，本章我们会讲述请求-应答模式的可靠性问题，并使用ZMQ提供的套接字类型组建起可靠的请求-应答消息系统。

本章将介绍的内容有：

  • 客户端请求-应答
  • 最近最少使用队列
  • 心跳机制
  • 面向服务的队列
  • 基于磁盘（脱机）队列
  • 主从备份服务
  • 无中间件的请求-应答

什么是可靠性？

要给可靠性下定义，我们可以先界定它的相反面——故障。如果我们可以处理某些类型的故障，那么我们的模型对于这些故障就是可靠的。下面我们就来列举分布式ZMQ应用程序中可能发生的问题，从可能性高的故障开始：

  • 应用程序代码是最大的故障来源。程序会崩溃或中止，停止对数据来源的响应，或是响应得太慢，耗尽内存等。
  • 系统代码，如使用ZMQ编写的中间件，也会意外中止。系统代码应该要比应用程序代码更为可靠，但毕竟也有可能崩溃。特别是当系统代码与速度过慢的客户端交互时，很容易耗尽内存。
  • 消息队列溢出，典型的情况是系统代码中没有对蛮客户端做积极的处理，任由消息队列溢出。
  • 网络临时中断，造成消息丢失。这类错误ZMQ应用程序是无法及时发现的，因为ZMQ会自动进行重连。
  • 硬件系统崩溃，导致所有进程中止。
  • 网络会出现特殊情形的中断，如交换机的某个端口发生故障，导致部分网络无法访问。
  • 数据中心可能遭受雷击、地震、火灾、电压过载、冷却系统失效等。

想要让软件系统规避上述所有的风险，需要大量的人力物力，故不在本指南的讨论范围之内。

由于前五个故障类型涵盖了99.9%的情形（这一数据源自我近期进行的一项研究），所以我们会深入探讨。如果你的公司大到足以考虑最后两种情形，那请及时联系我，因为我正愁没钱将我家后院的大坑建成游泳池。

可靠性设计

简单地来说，可靠性就是当程序发生故障时也能顺利地运行下去，这要比搭建一个消息系统来得困难得多。我们会根据ZMQ提供的每一种核心消息模式，来看看如何保障代码的持续运行。

  • 请求-应答模式：当服务端在处理请求是中断，客户端能够得知这一信息，并停止接收消息，转而选择等待重试、请求另一服务端等操作。这里我们暂不讨论客户端发生问题的情形。

  • 发布-订阅模式：如果客户端收到一些消息后意外中止，服务端是不知道这一情况的。发布-订阅模式中的订阅者不会返回任何消息给发布者。但是，订阅者可以通过其他方式联系服务端，如请求-应答模式，要求服务端重发消息。这里我们暂不讨论服务端发生问题的情形。此外，订阅者可以通过某些方式检查自身是否运行得过慢，并采取相应措施（向操作者发出警告、中止等）。

  • 管道模式：如果worker意外终止，任务分发器将无从得知。管道模式和发布-订阅模式类似，只朝一个方向发送消息。但是，下游的结果收集器可以检测哪项任务没有完成，并告诉任务分发器重新分配该任务。如果任务分发器或结果收集器意外中止了，那客户端发出的请求只能另作处理。所以说，系统代码真的要减少出错的几率，因为这很难处理。

本章主要讲解请求-应答模式中的可靠性设计，其他模式将在后续章节中讲解。

最基本的请求应答模式是REQ客户端发送一个同步的请求至REP服务端，这种模式的可靠性很低。如果服务端在处理请求时中止，那客户端会永远处于等待状态。

相比TCP协议，ZMQ提供了自动重连机制、消息分发的负载均衡等。但是，在真实环境中这也是不够的。唯一可以完全信任基本请求-应答模式的应用场景是同一进程的两个线程之间进行通信，没有网络问题或服务器失效的情况。

但是，只要稍加修饰，这种基本的请求-应答模式就能很好地在现实环境中工作了。我喜欢将其称为“海盗”模式。

粗略地讲，客户端连接服务端有三种方式，每种方式都需要不同的可靠性设计：

  • 多个客户端直接和单个服务端进行通信。使用场景：只有一个单点服务器，所有客户端都需要和它通信。需处理的故障：服务器崩溃和重启；网络连接中断。

  • 多个客户端和单个队列装置通信，该装置将请求分发给多个服务端。使用场景：任务分发。需处理的故障：worker崩溃和重启，死循环，过载；队列装置崩溃和重启；网络中断。

  • 多个客户端直接和多个服务端通信，无中间件。使用场景：类似域名解析的分布式服务。需处理的故障：服务端崩溃和重启，死循环，过载；网络连接中断。

以上每种设计都必须有所取舍，很多时候会混合使用。下面我们详细说明。

客户端的可靠性设计（懒惰海盗模式）

我们可以通过在客户端进行简单的设置，来实现可靠的请求-应答模式。我暂且称之为“懒惰的海盗”（Lazy Pirate）模式。

在接收应答时，我们不进行同步等待，而是做以下操作：

  • 对REQ套接字进行轮询，当消息抵达时才进行接收；
  • 请求超时后重发消息，循环多次；
  • 若仍无消息，则结束当前事务。

1

使用REQ套接字时必须严格遵守发送-接收过程，因为它内部采用了一个有限状态机来限定状态，这一特性会让我们应用“海盗”模式时遇上一些麻烦。最简单的做法是将REQ套接字关闭重启，从而打破这一限定。

lpclient: Lazy Pirate client in C

  1. //
  2. //  Lazy Pirate client
  3. //  使用zmq_poll轮询来实现安全的请求-应答
  4. //  运行时可随机关闭或重启lpserver程序
  5. //
  6. #include "czmq.h"
  8. #define REQUEST_TIMEOUT     2500    //  毫秒, (> 1000!)
  9. #define REQUEST_RETRIES     3       //  尝试次数
  10. #define SERVER_ENDPOINT     "tcp://localhost:5555"
  12. int main (void)
  13. {
  14.     zctx_t *ctx = zctx_new ();
  15.     printf ("I: 正在连接服务器...\n");
  16.     void *client = zsocket_new (ctx, ZMQ_REQ);
  17.     assert (client);
  18.     zsocket_connect (client, SERVER_ENDPOINT);
  20.     int sequence = 0;
  21.     int retries_left = REQUEST_RETRIES;
  22.     while (retries_left && !zctx_interrupted) {
  23.         //  发送一个请求，并开始接收消息
  24.         char request [10];
  25.         sprintf (request, "%d", ++sequence);
  26.         zstr_send (client, request);
  28.         int expect_reply = 1;
  29.         while (expect_reply) {
  30.             //  对套接字进行轮询，并设置超时时间
  31.             zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } };
  32.             int rc = zmq_poll (items, 1, REQUEST_TIMEOUT * ZMQ_POLL_MSEC);
  33.             if (rc == -1)
  34.                 break;          //  中断
  36.             //  如果接收到回复则进行处理
  37.             if (items [0].revents & ZMQ_POLLIN) {
  38.                 //  收到服务器应答，必须和请求时的序号一致
  39.                 char *reply = zstr_recv (client);
  40.                 if (!reply)
  41.                     break;      //  Interrupted
  42.                 if (atoi (reply) == sequence) {
  43.                     printf ("I: 服务器返回正常 (%s)\n", reply);
  44.                     retries_left = REQUEST_RETRIES;
  45.                     expect_reply = 0;
  46.                 }
  47.                 else
  48.                     printf ("E: 服务器返回异常: %s\n",
  49.                         reply);
  51.                 free (reply);
  52.             }
  53.             else
  54.             if (--retries_left == 0) {
  55.                 printf ("E: 服务器不可用，取消操作\n");
  56.                 break;
  57.             }
  58.             else {
  59.                 printf ("W: 服务器没有响应，正在重试...\n");
  60.                 //  关闭旧套接字，并建立新套接字
  61.                 zsocket_destroy (ctx, client);
  62.                 printf ("I: 服务器重连中...\n");
  63.                 client = zsocket_new (ctx, ZMQ_REQ);
  64.                 zsocket_connect (client, SERVER_ENDPOINT);
  65.                 //  使用新套接字再次发送请求
  66.                 zstr_send (client, request);
  67.             }
  68.         }
  69.     }
  70.     zctx_destroy (&ctx);
  71.     return 0;
  72. }

lpserver: Lazy Pirate server in C

  1. //
  2. //  Lazy Pirate server
  3. //  将REQ套接字连接至 tcp://*:5555
  4. //  和hwserver程序类似，除了以下两点：
  5. //   - 直接输出请求内容
  6. //   - 随机地降慢运行速度，或中止程序，模拟崩溃
  7. //
  8. #include "zhelpers.h"
  10. int main (void)
  11. {
  12.     srandom ((unsigned) time (NULL));
  14.     void *context = zmq_init (1);
  15.     void *server = zmq_socket (context, ZMQ_REP);
  16.     zmq_bind (server, "tcp://*:5555");
  18.     int cycles = 0;
  19.     while (1) {
  20.         char *request = s_recv (server);
  21.         cycles++;
  23.         //  循环几次后开始模拟各种故障
  24.         if (cycles > 3 && randof (3) == 0) {
  25.             printf ("I: 模拟程序崩溃\n");
  26.             break;
  27.         }
  28.         else
  29.         if (cycles > 3 && randof (3) == 0) {
  30.             printf ("I: 模拟CPU过载\n");
  31.             sleep (2);
  32.         }
  33.         printf ("I: 正常请求 (%s)\n", request);
  34.         sleep (1);              //  耗时的处理过程
  35.         s_send (server, request);
  36.         free (request);
  37.     }
  38.     zmq_close (server);
  39.     zmq_term (context);
  40.     return 0;
  41. }

运行这个测试用例时，可以打开两个控制台，服务端会随机发生故障，你可以看看客户端的反应。服务端的典型输出如下：

  1. I: normal request (1)
  2. I: normal request (2)
  3. I: normal request (3)
  4. I: simulating CPU overload
  5. I: normal request (4)
  6. I: simulating a crash

客户端的输出是：

  1. I: connecting to server...
  2. I: server replied OK (1)
  3. I: server replied OK (2)
  4. I: server replied OK (3)
  5. W: no response from server, retrying...
  6. I: connecting to server...
  7. W: no response from server, retrying...
  8. I: connecting to server...
  9. E: server seems to be offline, abandoning

客户端为每次请求都加上了序列号，并检查收到的应答是否和序列号一致，以保证没有请求或应答丢失，同一个应答收到多次或乱序。多运行几次实例，看看是否真的能够解决问题。现实环境中你不需要使用到序列号，那只是为了证明这一方式是可行的。

客户端使用REQ套接字进行请求，并在发生问题时打开一个新的套接字来，绕过REQ强制的发送/接收过程。可能你会想用DEALER套接字，但这并不是一个好主意。首先，DEALER并不会像REQ那样处理信封（如果你不知道信封是什么，那更不能用DEALER了）。其次，你可能会获得你并不想得到的结果。

这一方案的优劣是：

  • 优点：简单明了，容易实施；
  • 优点：可以方便地应用到现有的客户端和服务端程序中；
  • 优点：ZMQ有自动重连机制；
  • 缺点：单点服务发生故障时不能定位到新的可用服务。

基本的可靠队列（简单海盗模式）

在第二种模式中，我们使用一个队列装置来扩展上述的“懒惰的海盗”模式，使客户端能够透明地和多个服务端通信。这里的服务端可以定义为worker。我们可以从最基础的模型开始，分阶段实施这个方案。

在所有的海盗模式中，worker是无状态的，或者说存在着一个我们所不知道的公共状态，如共享数据库。队列装置的存在意味着worker可以在client毫不知情的情况下随意进出。一个worker死亡后，会有另一个worker接替它的工作。这种拓扑结果非常简洁，但唯一的缺点是队列装置本身会难以维护，可能造成单点故障。

在第三章中，队列装置的基本算法是最近最少使用算法。那么，如果worker死亡或阻塞，我们需要做些什么？答案是很少很少。我们已经在client中加入了重试的机制，所以，使用基本的LRU队列就可以运作得很好了。这种做法也符合ZMQ的逻辑，所以我们可以通过在点对点交互中插入一个简单的队列装置来扩展它：

2

我们可以直接使用“懒惰的海盗”模式中的client，以下是队列装置的代码：

spqueue: Simple Pirate queue in C

  1. //
  2. //  简单海盗队列
  3. //  
  4. //  这个装置和LRU队列完全一致，不存在任何可靠性机制，依靠client的重试来保证装置的运行
  5. //
  6. #include "czmq.h"
  8. #define LRU_READY   "\001"      //  消息：worker准备就绪
  10. int main (void)
  11. {
  12.     //  准备上下文和套接字
  13.     zctx_t *ctx = zctx_new ();
  14.     void *frontend = zsocket_new (ctx, ZMQ_ROUTER);
  15.     void *backend = zsocket_new (ctx, ZMQ_ROUTER);
  16.     zsocket_bind (frontend, "tcp://*:5555");    //  client端点
  17.     zsocket_bind (backend,  "tcp://*:5556");    //  worker端点
  19.     //  存放可用worker的队列
  20.     zlist_t *workers = zlist_new ();
  22.     while (1) {
  23.         zmq_pollitem_t items [] = {
  24.             { backend,  0, ZMQ_POLLIN, 0 },
  25.             { frontend, 0, ZMQ_POLLIN, 0 }
  26.         };
  27.         //  当有可用的woker时，轮询前端端点
  28.         int rc = zmq_poll (items, zlist_size (workers)? 2: 1, -1);
  29.         if (rc == -1)
  30.             break;              //  中断
  32.         //  处理后端端点的worker消息
  33.         if (items [0].revents & ZMQ_POLLIN) {
  34.             //  使用worker的地址进行LRU排队
  35.             zmsg_t *msg = zmsg_recv (backend);
  36.             if (!msg)
  37.                 break;          //  中断
  38.             zframe_t *address = zmsg_unwrap (msg);
  39.             zlist_append (workers, address);
  41.             //  如果消息不是READY，则转发给client
  42.             zframe_t *frame = zmsg_first (msg);
  43.             if (memcmp (zframe_data (frame), LRU_READY, 1) == 0)
  44.                 zmsg_destroy (&msg);
  45.             else
  46.                 zmsg_send (&msg, frontend);
  47.         }
  48.         if (items [1].revents & ZMQ_POLLIN) {
  49.             //  获取client请求，转发给第一个可用的worker
  50.             zmsg_t *msg = zmsg_recv (frontend);
  51.             if (msg) {
  52.                 zmsg_wrap (msg, (zframe_t *) zlist_pop (workers));
  53.                 zmsg_send (&msg, backend);
  54.             }
  55.         }
  56.     }
  57.     //  程序运行结束，进行清理
  58.     while (zlist_size (workers)) {
  59.         zframe_t *frame = (zframe_t *) zlist_pop (workers);
  60.         zframe_destroy (&frame);
  61.     }
  62.     zlist_destroy (&workers);
  63.     zctx_destroy (&ctx);
  64.     return 0;
  65. }

以下是worker的代码，用到了“懒惰的海盗”服务，并将其调整为LRU模式（使用REQ套接字传递“已就绪”信号）：

spworker: Simple Pirate worker in C

  1. //
  2. //  简单海盗模式worker
  3. //  
  4. //  使用REQ套接字连接tcp://*:5556，使用LRU算法实现worker
  5. //
  6. #include "czmq.h"
  7. #define LRU_READY   "\001"      //  消息：worker已就绪
  9. int main (void)
  10. {
  11.     zctx_t *ctx = zctx_new ();
  12.     void *worker = zsocket_new (ctx, ZMQ_REQ);
  14.     //  使用随机符号来指定套接字标识，方便追踪
  15.     srandom ((unsigned) time (NULL));
  16.     char identity [10];
  17.     sprintf (identity, "%04X-%04X", randof (0x10000), randof (0x10000));
  18.     zmq_setsockopt (worker, ZMQ_IDENTITY, identity, strlen (identity));
  19.     zsocket_connect (worker, "tcp://localhost:5556");
  21.     //  告诉代理worker已就绪
  22.     printf ("I: (%s) worker准备就绪\n", identity);
  23.     zframe_t *frame = zframe_new (LRU_READY, 1);
  24.     zframe_send (&frame, worker, 0);
  26.     int cycles = 0;
  27.     while (1) {
  28.         zmsg_t *msg = zmsg_recv (worker);
  29.         if (!msg)
  30.             break;              //  中断
  32.         //  经过几轮循环后，模拟各种问题
  33.         cycles++;
  34.         if (cycles > 3 && randof (5) == 0) {
  35.             printf ("I: (%s) 模拟崩溃\n", identity);
  36.             zmsg_destroy (&msg);
  37.             break;
  38.         }
  39.         else
  40.         if (cycles > 3 && randof (5) == 0) {
  41.             printf ("I: (%s) 模拟CPU过载\n", identity);
  42.             sleep (3);
  43.             if (zctx_interrupted)
  44.                 break;
  45.         }
  46.         printf ("I: (%s) 正常应答\n", identity);
  47.         sleep (1);              //  进行某些处理
  48.         zmsg_send (&msg, worker);
  49.     }
  50.     zctx_destroy (&ctx);
  51.     return 0;
  52. }

运行上述事例，启动多个worker，一个client，以及一个队列装置，顺序随意。你可以看到worker最终都会崩溃或死亡，client则多次重试并最终放弃。装置从来不会停止，你可以任意重启worker和client，这个模型可以和任意个worker、client交互。

健壮的可靠队列（偏执海盗模式）

“简单海盗队列”模式工作得非常好，主要是因为它只是两个现有模式的结合体。不过，它也有一些缺点：

  • 该模式无法处理队列的崩溃或重启。client会进行重试，但worker不会重启。虽然ZMQ会自动重连worker的套接字，但对于新启动的队列装置来说，由于worker并没有发送“已就绪”的消息，所以它相当于是不存在的。为了解决这一问题，我们需要从队列发送心跳给worker，这样worker就能知道队列是否已经死亡。

  • 队列没有检测worker是否已经死亡，所以当worker在处于空闲状态时死亡，队列装置只有在发送了某个请求之后才会将该worker从队列中移除。这时，client什么都不能做，只能等待。这不是一个致命的问题，但是依然是不够好的。所以，我们需要从worker发送心跳给队列装置，从而让队列得知worker什么时候消亡。

我们使用一个名为“偏执的海盗模式”来解决上述两个问题。

之前我们使用REQ套接字作为worker的套接字类型，但在偏执海盗模式中我们会改用DEALER套接字，从而使我们能够任意地发送和接受消息，而不是像REQ套接字那样必须完成发送-接受循环。而DEALER的缺点是我们必须自己管理消息信封。如果你不知道信封是什么，那请阅读第三章。

3

我们仍会使用懒惰海盗模式的client，以下是偏执海盗的队列装置代码：

ppqueue: Paranoid Pirate queue in C

  1. //
  2. //  偏执海盗队列
  3. //
  4. #include "czmq.h"
  6. #define HEARTBEAT_LIVENESS  3       //  心跳健康度，3-5是合理的
  7. #define HEARTBEAT_INTERVAL  1000    //  单位：毫秒
  9. //  偏执海盗协议的消息代码
  10. #define PPP_READY       "\001"      //  worker已就绪
  11. #define PPP_HEARTBEAT   "\002"      //  worker心跳
  14. //  使用以下结构表示worker队列中的一个有效的worker
  16. typedef struct {
  17.     zframe_t *address;          //  worker的地址
  18.     char *identity;             //  可打印的套接字标识
  19.     int64_t expiry;             //  过期时间
  20. } worker_t;
  22. //  创建新的worker
  23. static worker_t *
  24. s_worker_new (zframe_t *address)
  25. {
  26.     worker_t *self = (worker_t *) zmalloc (sizeof (worker_t));
  27.     self->address = address;
  28.     self->identity = zframe_strdup (address);
  29.     self->expiry = zclock_time () + HEARTBEAT_INTERVAL * HEARTBEAT_LIVENESS;
  30.     return self;
  31. }
  33. //  销毁worker结构，包括标识
  34. static void
  35. s_worker_destroy (worker_t **self_p)
  36. {
  37.     assert (self_p);
  38.     if (*self_p) {
  39.         worker_t *self = *self_p;
  40.         zframe_destroy (&self->address);
  41.         free (self->identity);
  42.         free (self);
  43.         *self_p = NULL;
  44.     }
  45. }
  47. //  worker已就绪，将其移至列表末尾
  48. static void
  49. s_worker_ready (worker_t *self, zlist_t *workers)
  50. {
  51.     worker_t *worker = (worker_t *) zlist_first (workers);
  52.     while (worker) {
  53.         if (streq (self->identity, worker->identity)) {
  54.             zlist_remove (workers, worker);
  55.             s_worker_destroy (&worker);
  56.             break;
  57.         }
  58.         worker = (worker_t *) zlist_next (workers);
  59.     }
  60.     zlist_append (workers, self);
  61. }
  63. //  返回下一个可用的worker地址
  64. static zframe_t *
  65. s_workers_next (zlist_t *workers)
  66. {
  67.     worker_t *worker = zlist_pop (workers);
  68.     assert (worker);
  69.     zframe_t *frame = worker->address;
  70.     worker->address = NULL;
  71.     s_worker_destroy (&worker);
  72.     return frame;
  73. }
  75. //  寻找并销毁已过期的worker。
  76. //  由于列表中最旧的worker排在最前，所以当找到第一个未过期的worker时就停止。
  77. static void
  78. s_workers_purge (zlist_t *workers)
  79. {
  80.     worker_t *worker = (worker_t *) zlist_first (workers);
  81.     while (worker) {
  82.         if (zclock_time () < worker->expiry)
  83.             break;              //  worker未过期，停止扫描
  85.         zlist_remove (workers, worker);
  86.         s_worker_destroy (&worker);
  87.         worker = (worker_t *) zlist_first (workers);
  88.     }
  89. }
  92. int main (void)
  93. {
  94.     zctx_t *ctx = zctx_new ();
  95.     void *frontend = zsocket_new (ctx, ZMQ_ROUTER);
  96.     void *backend  = zsocket_new (ctx, ZMQ_ROUTER);
  97.     zsocket_bind (frontend, "tcp://*:5555");    //  client端点
  98.     zsocket_bind (backend,  "tcp://*:5556");    //  worker端点
  99.     //  List of available workers
  100.     zlist_t *workers = zlist_new ();
  102.     //  规律地发送心跳
  103.     uint64_t heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;
  105.     while (1) {
  106.         zmq_pollitem_t items [] = {
  107.             { backend,  0, ZMQ_POLLIN, 0 },
  108.             { frontend, 0, ZMQ_POLLIN, 0 }
  109.         };
  110.         //  当存在可用worker时轮询前端端点
  111.         int rc = zmq_poll (items, zlist_size (workers)? 2: 1,
  112.             HEARTBEAT_INTERVAL * ZMQ_POLL_MSEC);
  113.         if (rc == -1)
  114.             break;              //  中断
  116.         //  处理后端worker请求
  117.         if (items [0].revents & ZMQ_POLLIN) {
  118.             //  使用worker地址进行LRU路由
  119.             zmsg_t *msg = zmsg_recv (backend);
  120.             if (!msg)
  121.                 break;          //  中断
  123.             //  worker的任何信号均表示其仍然存活
  124.             zframe_t *address = zmsg_unwrap (msg);
  125.             worker_t *worker = s_worker_new (address);
  126.             s_worker_ready (worker, workers);
  128.             //  处理控制消息，或者将应答转发给client
  129.             if (zmsg_size (msg) == 1) {
  130.                 zframe_t *frame = zmsg_first (msg);
  131.                 if (memcmp (zframe_data (frame), PPP_READY, 1)
  132.                 &&  memcmp (zframe_data (frame), PPP_HEARTBEAT, 1)) {
  133.                     printf ("E: invalid message from worker");
  134.                     zmsg_dump (msg);
  135.                 }
  136.                 zmsg_destroy (&msg);
  137.             }
  138.             else
  139.                 zmsg_send (&msg, frontend);
  140.         }
  141.         if (items [1].revents & ZMQ_POLLIN) {
  142.             //  获取下一个client请求，交给下一个可用的worker
  143.             zmsg_t *msg = zmsg_recv (frontend);
  144.             if (!msg)
  145.                 break;          //  中断
  146.             zmsg_push (msg, s_workers_next (workers));
  147.             zmsg_send (&msg, backend);
  148.         }
  150.         //  发送心跳给空闲的worker
  151.         if (zclock_time () >= heartbeat_at) {
  152.             worker_t *worker = (worker_t *) zlist_first (workers);
  153.             while (worker) {
  154.                 zframe_send (&worker->address, backend,
  155.                              ZFRAME_REUSE + ZFRAME_MORE);
  156.                 zframe_t *frame = zframe_new (PPP_HEARTBEAT, 1);
  157.                 zframe_send (&frame, backend, 0);
  158.                 worker = (worker_t *) zlist_next (workers);
  159.             }
  160.             heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;
  161.         }
  162.         s_workers_purge (workers);
  163.     }
  165.     //  程序结束后进行清理
  166.     while (zlist_size (workers)) {
  167.         worker_t *worker = (worker_t *) zlist_pop (workers);
  168.         s_worker_destroy (&worker);
  169.     }
  170.     zlist_destroy (&workers);
  171.     zctx_destroy (&ctx);
  172.     return 0;
  173. }

该队列装置使用心跳机制扩展了LRU模式，看起来很简单，但要想出这个主意还挺难的。下文会更多地介绍心跳机制。

以下是偏执海盗的worker代码：

ppworker: Paranoid Pirate worker in C

  1. //
  2. //  偏执海盗worker
  3. //
  4. #include "czmq.h"
  6. #define HEARTBEAT_LIVENESS  3       //  合理值：3-5
  7. #define HEARTBEAT_INTERVAL  1000    //  单位：毫秒
  8. #define INTERVAL_INIT       1000    //  重试间隔
  9. #define INTERVAL_MAX       32000    //  回退算法最大值
  11. //  偏执海盗规范的常量定义
  12. #define PPP_READY       "\001"      //  消息：worker已就绪
  13. #define PPP_HEARTBEAT   "\002"      //  消息：worker心跳
  15. //  返回一个连接至偏执海盗队列装置的套接字
  17. static void *
  18. s_worker_socket (zctx_t *ctx) {
  19.     void *worker = zsocket_new (ctx, ZMQ_DEALER);
  20.     zsocket_connect (worker, "tcp://localhost:5556");
  22.     //  告知队列worker已准备就绪
  23.     printf ("I: worker已就绪\n");
  24.     zframe_t *frame = zframe_new (PPP_READY, 1);
  25.     zframe_send (&frame, worker, 0);
  27.     return worker;
  28. }
  30. int main (void)
  31. {
  32.     zctx_t *ctx = zctx_new ();
  33.     void *worker = s_worker_socket (ctx);
  35.     //  如果心跳健康度为零，则表示队列装置已死亡
  36.     size_t liveness = HEARTBEAT_LIVENESS;
  37.     size_t interval = INTERVAL_INIT;
  39.     //  规律地发送心跳
  40.     uint64_t heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;
  42.     srandom ((unsigned) time (NULL));
  43.     int cycles = 0;
  44.     while (1) {
  45.         zmq_pollitem_t items [] = { { worker,  0, ZMQ_POLLIN, 0 } };
  46.         int rc = zmq_poll (items, 1, HEARTBEAT_INTERVAL * ZMQ_POLL_MSEC);
  47.         if (rc == -1)
  48.             break;              //  中断
  50.         if (items [0].revents & ZMQ_POLLIN) {
  51.             //  获取消息
  52.             //  - 3段消息，信封+内容，表示一个请求
  53.             //  - 1段消息，表示心跳
  54.             zmsg_t *msg = zmsg_recv (worker);
  55.             if (!msg)
  56.                 break;          //  中断
  58.             if (zmsg_size (msg) == 3) {
  59.                 //  若干词循环后模拟各种问题
  60.                 cycles++;
  61.                 if (cycles > 3 && randof (5) == 0) {
  62.                     printf ("I: 模拟崩溃\n");
  63.                     zmsg_destroy (&msg);
  64.                     break;
  65.                 }
  66.                 else
  67.                 if (cycles > 3 && randof (5) == 0) {
  68.                     printf ("I: 模拟CPU过载\n");
  69.                     sleep (3);
  70.                     if (zctx_interrupted)
  71.                         break;
  72.                 }
  73.                 printf ("I: 正常应答\n");
  74.                 zmsg_send (&msg, worker);
  75.                 liveness = HEARTBEAT_LIVENESS;
  76.                 sleep (1);              //  做一些处理工作
  77.                 if (zctx_interrupted)
  78.                     break;
  79.             }
  80.             else
  81.             if (zmsg_size (msg) == 1) {
  82.                 zframe_t *frame = zmsg_first (msg);
  83.                 if (memcmp (zframe_data (frame), PPP_HEARTBEAT, 1) == 0)
  84.                     liveness = HEARTBEAT_LIVENESS;
  85.                 else {
  86.                     printf ("E: 非法消息\n");
  87.                     zmsg_dump (msg);
  88.                 }
  89.                 zmsg_destroy (&msg);
  90.             }
  91.             else {
  92.                 printf ("E: 非法消息\n");
  93.                 zmsg_dump (msg);
  94.             }
  95.             interval = INTERVAL_INIT;
  96.         }
  97.         else
  98.         if (--liveness == 0) {
  99.             printf ("W: 心跳失败，无法连接队列装置\n");
  100.             printf ("W: %zd 毫秒后进行重连...\n", interval);
  101.             zclock_sleep (interval);
  103.             if (interval < INTERVAL_MAX)
  104.                 interval *= 2;
  105.             zsocket_destroy (ctx, worker);
  106.             worker = s_worker_socket (ctx);
  107.             liveness = HEARTBEAT_LIVENESS;
  108.         }
  110.         //  适时发送心跳给队列
  111.         if (zclock_time () > heartbeat_at) {
  112.             heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;
  113.             printf ("I: worker心跳\n");
  114.             zframe_t *frame = zframe_new (PPP_HEARTBEAT, 1);
  115.             zframe_send (&frame, worker, 0);
  116.         }
  117.     }
  118.     zctx_destroy (&ctx);
  119.     return 0;
  120. }

几点说明：

  • 代码中包含了几处失败模拟，和先前一样。这会让代码极难维护，所以当投入使用时，应当移除这些模拟代码。
  • 偏执海盗模式中队列的心跳有时会不正常，下文会讲述这一点。
  • worker使用了一种类似于懒惰海盗client的重试机制，但有两点不同：1、回退算法设置；2、永不言弃。

尝试运行以下代码，跑通流程：

  1. ppqueue &
  2. for i in 1 2 3 4; do
  3.     ppworker &
  4.     sleep 1
  5. done
  6. lpclient &

你会看到worker逐个崩溃，client在多次尝试后放弃。你可以停止并重启队列装置，client和worker会相继重连，并正确地发送、处理和接收请求，顺序不会混乱。所以说，整个通信过程只有两种情形：交互成功，或client最终放弃。

心跳

当我在写偏执海盗模式的示例时，大约花了五个小时的时间来协调队列至worker的心跳，剩下的请求-应答链路只花了约10分钟的时间。心跳机制在可靠性上带来的益处有时还不及它所引发的问题。使用过程中很有可能会产生“虚假故障”的情况，即节点误认为他们已失去连接，因为心跳没有正确地发送。

在理解和实施心跳时，需要考虑以下几点：

  • 心跳不是一种请求-应答，它们异步地在节点之间传递，任一节点都可以通过它来判断对方已经死亡，并中止通信。

  • 如果某个节点使用持久套接字（即设定了套接字标识），意味着发送给它的心跳可能会堆砌，并在重连后一起收到。所以说，worker不应该使用持久套接字。示例代码使用持久套接字是为了便于调试，而且代码中使用了随机的套接字标识，避免重用之前的标识。

  • 使用过程中，应先让心跳工作起来，再进行后面的消息处理。你需要保证启动任一节点后，心跳都能正确地执行。停止并重启他们，模拟冻结、崩溃等情况来进行测试。

  • 当你的主循环使用了zmq_poll()，则应该使用另一个计时器来触发心跳。不要使用主循环来控制心跳的发送，这回导致过量地发送心跳（阻塞网络），或是发送得太少（导致节点断开）。zhelpers包提供了s_clock()函数返回当前系统时间戳，单位是毫秒，可以用它来控制心跳的发送间隔。C代码如下：

  1. // 规律地发送心跳
  2. uint64_t heartbeat_at = s_clock () + HEARTBEAT_INTERVAL;
  3. while (1) {
  4.     …
  5.     zmq_poll (items, 1, HEARTBEAT_INTERVAL * 1000);
  6.     …
  7.     // 无论zmq_poll的行为是什么，都使用以下逻辑判断是否发送心跳
  8.     if (s_clock () > heartbeat_at) {
  9.         … 发送心跳给所有节点
  10.         // 设置下一次心跳的时间
  11.         heartbeat_at = s_clock () + HEARTBEAT_INTERVAL;
  12.     }
  13. }

  • 主循环应该使用心跳间隔作为超时时间。显然不能使用无超时时间的设置，而短于心跳间隔也只是浪费循环次数而已。

  • 使用简单的追踪方式来进行追踪，如直接输出至控制台。这里有一些追踪的窍门：使用zmsg()函数打印套接字内容；对消息进行编号，判断是否会有间隔。

  • 在真实的应用程序中，心跳必须是可以配置的，并能和节点共同商定。有些节点需要高频心跳，如10毫秒，另一些节点则可能只需要30秒发送一次心跳即可。

  • 如果你要对不同的节点发送不同频率的心跳，那么poll的超时时间应设置为最短的心跳间隔。

  • 也许你会想要用一个单独的套接字来处理心跳，这看起来很棒，可以将同步的请求-应答和异步的心跳隔离开来。但是，这个主意并不好，原因有几点：首先、发送数据时其实是不需要发送心跳的；其次、套接字可能会因为网络问题而阻塞，你需要设法知道用于发送数据的套接字停止响应的原因是死亡了还是过于繁忙而已，这样你就需要对这个套接字进行心跳。最后，处理两个套接字要比处理一个复杂得多。

  • 我们没有设置client至队列的心跳，因为这太过复杂了，而且没有太大价值。

约定和协议

也许你已经注意到，由于心跳机制，偏执海盗模式和简单海盗模式是不兼容的。

其实，这里我们需要写一个协议。也许在试验阶段是不需要协议的，但这在真实的应用程序中是有必要。如果我们想用其他语言来写worker怎么办？我们是否需要通过源代码来查看通信过程？如果我们想改变协议怎么办？规范可能很简单，但并不显然。越是成功的协议，就会越为复杂。

一个缺乏约定的应用程序一定是不可复用的，所以让我们来为这个协议写一个规范，怎么做呢？

  • 位于rfc.zeromq.org的wiki页上，我们特地设置了一个用于存放ZMQ协议的页面。
    • 要创建一个新的协议，你需要注册并按照指导进行。过程很直接，但并不一定所有人都能撰写技术性文档。

我大约花了15分钟的时间草拟海盗模式规范（PPP），麻雀虽小，但五脏俱全。

要用PPP协议进行真实环境下的编程，你还需要：

  • 在READY命令中加入版本号，这样就能再日后安全地新增PPP版本号。
  • 目前，READY和HEARTBEAT信号并没有指定其来源于请求还是应答。要区分他们，需要新建一个消息结构，其中包含“消息类型”这一信息。

面向服务的可靠队列（管家模式）

世上的事物往往瞬息万变，正当我们期待有更好的协议来解决上一节的问题时，已经有人制定好了：

这份协议只有一页，它将PPP协议变得更为坚固。我们在设计复杂架构时应该这样做：首先写下约定，再用软件去实现它。

管家模式协议（MDP）在扩展PPP协议时引入了一个有趣的特性：client发送的每一个请求都有一个“服务名称”，而worker在像队列装置注册时需要告知自己的服务类型。MDP的优势在于它来源于现实编程，协议简单，且容易提升。

引入“服务名称”的机制，是对偏执海盗队列的一个简单补充，而结果是让其成为一个面向服务的代理。

4

在实施管家模式之前，我们需要为client和worker编写一个框架。如果程序员可以通过简单的API来实现这种模式，那就没有必要让他们去了解管家模式的协议内容和实现方法了。
所以，我们第一个协议（即管家模式协议）定义了分布式架构中节点是如何互相交互的，第二个协议则要定义应用程序应该如何通过框架来使用这一协议。
管家模式有两个端点，客户端和服务端。因为我们要为client和worker都撰写框架，所以就需要提供两套API。以下是用简单的面向对象方法设计的client端API雏形，使用的是C语言的ZFL library。

  1. mdcli_t *mdcli_new     (char *broker);
  2. void     mdcli_destroy (mdcli_t **self_p);
  3. zmsg_t  *mdcli_send    (mdcli_t *self, char *service, zmsg_t **request_p);

就这么简单。我们创建了一个会话来和代理通信，发送并接收一个请求，最后关闭连接。以下是worker端API的雏形。

  1. mdwrk_t *mdwrk_new     (char *broker,char *service);
  2. void     mdwrk_destroy (mdwrk_t **self_p);
  3. zmsg_t  *mdwrk_recv    (mdwrk_t *self, zmsg_t *reply);

上面两段代码看起来差不多，但是worker端API略有不同。worker第一次执行recv()后会传递一个空的应答，之后才传递当前的应答，并获得新的请求。

两段的API都很容易开发，只需在偏执海盗模式代码的基础上修改即可。以下是client API：

mdcliapi: Majordomo client API in C

  1. /*  =====================================================================
  2.     mdcliapi.c
  4.     Majordomo Protocol Client API
  5.     Implements the MDP/Worker spec at http://rfc.zeromq.org/spec:7.
  7.     ---------------------------------------------------------------------
  8.     Copyright (c) 1991-2011 iMatix Corporation <www.imatix.com>
  9.     Copyright other contributors as noted in the AUTHORS file.
  11.     This file is part of the ZeroMQ Guide: http://zguide.zeromq.org
  13.     This is free software; you can redistribute it and/or modify it under
  14.     the terms of the GNU Lesser General Public License as published by
  15.     the Free Software Foundation; either version 3 of the License, or (at
  16.     your option) any later version.
  18.     This software is distributed in the hope that it will be useful, but
  19.     WITHOUT ANY WARRANTY; without even the implied warranty of
  20.     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  21.     Lesser General Public License for more details.
  23.     You should have received a copy of the GNU Lesser General Public
  24.     License along with this program. If not, see
  25.     <http://www.gnu.org/licenses/>.
  26.     =====================================================================
  27. */
  29. #include "mdcliapi.h"
  31. //  类结构
  32. //  我们会通过成员方法来访问这些属性
  34. struct _mdcli_t {
  35.     zctx_t *ctx;                //  上下文
  36.     char *broker;
  37.     void *client;               //  连接至代理的套接字
  38.     int verbose;                //  使用标准输出打印当前活动
  39.     int timeout;                //  请求超时时间
  40.     int retries;                //  请求重试次数
  41. };
  44. //  ---------------------------------------------------------------------
  45. //  连接或重连代理
  47. void s_mdcli_connect_to_broker (mdcli_t *self)
  48. {
  49.     if (self->client)
  50.         zsocket_destroy (self->ctx, self->client);
  51.     self->client = zsocket_new (self->ctx, ZMQ_REQ);
  52.     zmq_connect (self->client, self->broker);
  53.     if (self->verbose)
  54.         zclock_log ("I: 正在连接至代理 %s...", self->broker);
  55. }
  58. //  ---------------------------------------------------------------------
  59. //  构造函数
  61. mdcli_t *
  62. mdcli_new (char *broker, int verbose)
  63. {
  64.     assert (broker);
  66.     mdcli_t *self = (mdcli_t *) zmalloc (sizeof (mdcli_t));
  67.     self->ctx = zctx_new ();
  68.     self->broker = strdup (broker);
  69.     self->verbose = verbose;
  70.     self->timeout = 2500;           //  毫秒
  71.     self->retries = 3;              //  尝试次数
  73.     s_mdcli_connect_to_broker (self);
  74.     return self;
  75. }
  78. //  ---------------------------------------------------------------------
  79. //  析构函数
  81. void
  82. mdcli_destroy (mdcli_t **self_p)
  83. {
  84.     assert (self_p);
  85.     if (*self_p) {
  86.         mdcli_t *self = *self_p;
  87.         zctx_destroy (&self->ctx);
  88.         free (self->broker);
  89.         free (self);
  90.         *self_p = NULL;
  91.     }
  92. }
  95. //  ---------------------------------------------------------------------
  96. //  设定请求超时时间
  98. void
  99. mdcli_set_timeout (mdcli_t *self, int timeout)
  100. {
  101.     assert (self);
  102.     self->timeout = timeout;
  103. }
  106. //  ---------------------------------------------------------------------
  107. //  设定请求重试次数
  109. void
  110. mdcli_set_retries (mdcli_t *self, int retries)
  111. {
  112.     assert (self);
  113.     self->retries = retries;
  114. }
  117. //  ---------------------------------------------------------------------
  118. //  向代理发送请求，并尝试获取应答；
  119. //  对消息保持所有权，发送后销毁；
  120. //  返回应答消息，或NULL。
  122. zmsg_t *
  123. mdcli_send (mdcli_t *self, char *service, zmsg_t **request_p)
  124. {
  125.     assert (self);
  126.     assert (request_p);
  127.     zmsg_t *request = *request_p;
  129.     //  用协议前缀包装消息
  130.     //  Frame 1: "MDPCxy" (six bytes, MDP/Client x.y)
  131.     //  Frame 2: 服务名称 (可打印字符串)
  132.     zmsg_pushstr (request, service);
  133.     zmsg_pushstr (request, MDPC_CLIENT);
  134.     if (self->verbose) {
  135.         zclock_log ("I: 发送请求给 '%s' 服务:", service);
  136.         zmsg_dump (request);
  137.     }
  139.     int retries_left = self->retries;
  140.     while (retries_left && !zctx_interrupted) {
  141.         zmsg_t *msg = zmsg_dup (request);
  142.         zmsg_send (&msg, self->client);
  144.         while (TRUE) {
  145.             //  轮询套接字以接收应答，有超时时间
  146.             zmq_pollitem_t items [] = {
  147.                 { self->client, 0, ZMQ_POLLIN, 0 } };
  148.             int rc = zmq_poll (items, 1, self->timeout * ZMQ_POLL_MSEC);
  149.             if (rc == -1)
  150.                 break;          //  中断
  152.             //  收到应答后进行处理
  153.             if (items [0].revents & ZMQ_POLLIN) {
  154.                 zmsg_t *msg = zmsg_recv (self->client);
  155.                 if (self->verbose) {
  156.                     zclock_log ("I: received reply:");
  157.                     zmsg_dump (msg);
  158.                 }
  159.                 //  不要尝试处理错误，直接报错即可
  160.                 assert (zmsg_size (msg) >= 3);
  162.                 zframe_t *header = zmsg_pop (msg);
  163.                 assert (zframe_streq (header, MDPC_CLIENT));
  164.                 zframe_destroy (&header);
  166.                 zframe_t *reply_service = zmsg_pop (msg);
  167.                 assert (zframe_streq (reply_service, service));
  168.                 zframe_destroy (&reply_service);
  170.                 zmsg_destroy (&request);
  171.                 return msg;     //  成功
  172.             }
  173.             else
  174.             if (--retries_left) {
  175.                 if (self->verbose)
  176.                     zclock_log ("W: no reply, reconnecting...");
  177.                 //  重连并重发消息
  178.                 s_mdcli_connect_to_broker (self);
  179.                 zmsg_t *msg = zmsg_dup (request);
  180.                 zmsg_send (&msg, self->client);
  181.             }
  182.             else {
  183.                 if (self->verbose)
  184.                     zclock_log ("W: 发生严重错误，放弃重试。");
  185.                 break;          //  放弃
  186.             }
  187.         }
  188.     }
  189.     if (zctx_interrupted)
  190.         printf ("W: 收到中断消息，结束client进程...\n");
  191.     zmsg_destroy (&request);
  192.     return NULL;
  193. }

以下测试程序会执行10万次请求应答：

mdclient: Majordomo client application in C

  1. //
  2. //  管家模式协议 - 客户端示例
  3. //  使用mdcli API隐藏管家模式协议的内部实现
  4. //
  6. //  让我们直接编译这段代码，不生成类库
  7. #include "mdcliapi.c"
  9. int main (int argc, char *argv [])
  10. {
  11.     int verbose = (argc > 1 && streq (argv [1], "-v"));
  12.     mdcli_t *session = mdcli_new ("tcp://localhost:5555", verbose);
  14.     int count;
  15.     for (count = 0; count < 100000; count++) {
  16.         zmsg_t *request = zmsg_new ();
  17.         zmsg_pushstr (request, "Hello world");
  18.         zmsg_t *reply = mdcli_send (session, "echo", &request);
  19.         if (reply)
  20.             zmsg_destroy (&reply);
  21.         else
  22.             break;              //  中断或停止
  23.     }
  24.     printf ("已处理 %d 次请求-应答\n", count);
  25.     mdcli_destroy (&session);
  26.     return 0;
  27. }

下面是worker的API：

mdwrkapi: Majordomo worker API in C

  1. /*  =====================================================================
  2.     mdwrkapi.c
  4.     Majordomo Protocol Worker API
  5.     Implements the MDP/Worker spec at http://rfc.zeromq.org/spec:7.
  7.     ---------------------------------------------------------------------
  8.     Copyright (c) 1991-2011 iMatix Corporation <www.imatix.com>
  9.     Copyright other contributors as noted in the AUTHORS file.
  11.     This file is part of the ZeroMQ Guide: http://zguide.zeromq.org
  13.     This is free software; you can redistribute it and/or modify it under
  14.     the terms of the GNU Lesser General Public License as published by
  15.     the Free Software Foundation; either version 3 of the License, or (at
  16.     your option) any later version.
  18.     This software is distributed in the hope that it will be useful, but
  19.     WITHOUT ANY WARRANTY; without even the implied warranty of
  20.     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  21.     Lesser General Public License for more details.
  23.     You should have received a copy of the GNU Lesser General Public
  24.     License along with this program. If not, see
  25.     <http://www.gnu.org/licenses/>.
  26.     =====================================================================
  27. */
  29. #include "mdwrkapi.h"
  31. //  可靠性参数
  32. #define HEARTBEAT_LIVENESS  3       //  合理值：3-5
  34. //  类结构
  35. //  使用成员函数访问属性
  37. struct _mdwrk_t {
  38.     zctx_t *ctx;                //  上下文
  39.     char *broker;
  40.     char *service;
  41.     void *worker;               //  连接至代理的套接字
  42.     int verbose;                //  使用标准输出打印活动
  44.     //  心跳设置
  45.     uint64_t heartbeat_at;      //  发送心跳的时间
  46.     size_t liveness;            //  尝试次数
  47.     int heartbeat;              //  心跳延时，单位：毫秒
  48.     int reconnect;              //  重连延时，单位：毫秒
  50.     //  内部状态
  51.     int expect_reply;           //  初始值为0
  53.     //  应答地址，如果存在的话
  54.     zframe_t *reply_to;
  55. };
  58. //  ---------------------------------------------------------------------
  59. //  发送消息给代理
  60. //  如果没有提供消息，则内部创建一个
  62. static void
  63. s_mdwrk_send_to_broker (mdwrk_t *self, char *command, char *option,
  64.                         zmsg_t *msg)
  65. {
  66.     msg = msg? zmsg_dup (msg): zmsg_new ();
  68.     //  将协议信封压入消息顶部
  69.     if (option)
  70.         zmsg_pushstr (msg, option);
  71.     zmsg_pushstr (msg, command);
  72.     zmsg_pushstr (msg, MDPW_WORKER);
  73.     zmsg_pushstr (msg, "");
  75.     if (self->verbose) {
  76.         zclock_log ("I: sending %s to broker",
  77.             mdps_commands [(int) *command]);
  78.         zmsg_dump (msg);
  79.     }
  80.     zmsg_send (&msg, self->worker);
  81. }
  84. //  ---------------------------------------------------------------------
  85. //  连接或重连代理
  87. void s_mdwrk_connect_to_broker (mdwrk_t *self)
  88. {
  89.     if (self->worker)
  90.         zsocket_destroy (self->ctx, self->worker);
  91.     self->worker = zsocket_new (self->ctx, ZMQ_DEALER);
  92.     zmq_connect (self->worker, self->broker);
  93.     if (self->verbose)
  94.         zclock_log ("I: 正在连接代理 %s...", self->broker);
  96.     //  向代理注册服务类型
  97.     s_mdwrk_send_to_broker (self, MDPW_READY, self->service, NULL);
  99.     //  当心跳健康度为零，表示代理已断开连接
  100.     self->liveness = HEARTBEAT_LIVENESS;
  101.     self->heartbeat_at = zclock_time () + self->heartbeat;
  102. }
  105. //  ---------------------------------------------------------------------
  106. //  构造函数
  108. mdwrk_t *
  109. mdwrk_new (char *broker,char *service, int verbose)
  110. {
  111.     assert (broker);
  112.     assert (service);
  114.     mdwrk_t *self = (mdwrk_t *) zmalloc (sizeof (mdwrk_t));
  115.     self->ctx = zctx_new ();
  116.     self->broker = strdup (broker);
  117.     self->service = strdup (service);
  118.     self->verbose = verbose;
  119.     self->heartbeat = 2500;     //  毫秒
  120.     self->reconnect = 2500;     //  毫秒
  122.     s_mdwrk_connect_to_broker (self);
  123.     return self;
  124. }
  127. //  ---------------------------------------------------------------------
  128. //  析构函数
  130. void
  131. mdwrk_destroy (mdwrk_t **self_p)
  132. {
  133.     assert (self_p);
  134.     if (*self_p) {
  135.         mdwrk_t *self = *self_p;
  136.         zctx_destroy (&self->ctx);
  137.         free (self->broker);
  138.         free (self->service);
  139.         free (self);
  140.         *self_p = NULL;
  141.     }
  142. }
  145. //  ---------------------------------------------------------------------
  146. //  设置心跳延迟
  148. void
  149. mdwrk_set_heartbeat (mdwrk_t *self, int heartbeat)
  150. {
  151.     self->heartbeat = heartbeat;
  152. }
  155. //  ---------------------------------------------------------------------
  156. //  设置重连延迟
  158. void
  159. mdwrk_set_reconnect (mdwrk_t *self, int reconnect)
  160. {
  161.     self->reconnect = reconnect;
  162. }
  165. //  ---------------------------------------------------------------------
  166. //  若有应答则发送给代理，并等待新的请求
  168. zmsg_t *
  169. mdwrk_recv (mdwrk_t *self, zmsg_t **reply_p)
  170. {
  171.     //  格式化并发送请求传入的应答
  172.     assert (reply_p);
  173.     zmsg_t *reply = *reply_p;
  174.     assert (reply || !self->expect_reply);
  175.     if (reply) {
  176.         assert (self->reply_to);
  177.         zmsg_wrap (reply, self->reply_to);
  178.         s_mdwrk_send_to_broker (self, MDPW_REPLY, NULL, reply);
  179.         zmsg_destroy (reply_p);
  180.     }
  181.     self->expect_reply = 1;
  183.     while (TRUE) {
  184.         zmq_pollitem_t items [] = {
  185.             { self->worker,  0, ZMQ_POLLIN, 0 } };
  186.         int rc = zmq_poll (items, 1, self->heartbeat * ZMQ_POLL_MSEC);
  187.         if (rc == -1)
  188.             break;              //  中断
  190.         if (items [0].revents & ZMQ_POLLIN) {
  191.             zmsg_t *msg = zmsg_recv (self->worker);
  192.             if (!msg)
  193.                 break;          //  中断
  194.             if (self->verbose) {
  195.                 zclock_log ("I: 从代理处获得消息:");
  196.                 zmsg_dump (msg);
  197.             }
  198.             self->liveness = HEARTBEAT_LIVENESS;
  200.             //  不要处理错误，直接报错即可
  201.             assert (zmsg_size (msg) >= 3);
  203.             zframe_t *empty = zmsg_pop (msg);
  204.             assert (zframe_streq (empty, ""));
  205.             zframe_destroy (&empty);
  207.             zframe_t *header = zmsg_pop (msg);
  208.             assert (zframe_streq (header, MDPW_WORKER));
  209.             zframe_destroy (&header);
  211.             zframe_t *command = zmsg_pop (msg);
  212.             if (zframe_streq (command, MDPW_REQUEST)) {
  213.                 //  这里需要将消息中空帧之前的所有地址都保存起来，
  214.                 //  但在这里我们暂时只保存一个
  215.                 self->reply_to = zmsg_unwrap (msg);
  216.                 zframe_destroy (&command);
  217.                 return msg;     //  处理请求
  218.             }
  219.             else
  220.             if (zframe_streq (command, MDPW_HEARTBEAT))
  221.                 ;               //  不对心跳做任何处理
  222.             else
  223.             if (zframe_streq (command, MDPW_DISCONNECT))
  224.                 s_mdwrk_connect_to_broker (self);
  225.             else {
  226.                 zclock_log ("E: 消息不合法");
  227.                 zmsg_dump (msg);
  228.             }
  229.             zframe_destroy (&command);
  230.             zmsg_destroy (&msg);
  231.         }
  232.         else
  233.         if (--self->liveness == 0) {
  234.             if (self->verbose)
  235.                 zclock_log ("W: 失去与代理的连接 - 正在重试...");
  236.             zclock_sleep (self->reconnect);
  237.             s_mdwrk_connect_to_broker (self);
  238.         }
  239.         //  适时地发送消息
  240.         if (zclock_time () > self->heartbeat_at) {
  241.             s_mdwrk_send_to_broker (self, MDPW_HEARTBEAT, NULL, NULL);
  242.             self->heartbeat_at = zclock_time () + self->heartbeat;
  243.         }
  244.     }
  245.     if (zctx_interrupted)
  246.         printf ("W: 收到中断消息，中止worker...\n");
  247.     return NULL;
  248. }

以下测试程序实现了名为echo的服务：

mdworker: Majordomo worker application in C

  1. //
  2. //  管家模式协议 - worker示例
  3. //  使用mdwrk API隐藏MDP协议的内部实现
  4. //
  6. //  让我们直接编译代码，而不创建类库
  7. #include "mdwrkapi.c"
  9. int main (int argc, char *argv [])
  10. {
  11.     int verbose = (argc > 1 && streq (argv [1], "-v"));
  12.     mdwrk_t *session = mdwrk_new (
  13.         "tcp://localhost:5555", "echo", verbose);
  15.     zmsg_t *reply = NULL;
  16.     while (1) {
  17.         zmsg_t *request = mdwrk_recv (session, &reply);
  18.         if (request == NULL)
  19.             break;              //  worker被中止
  20.         reply = request;        //  echo服务……其实很复杂:)
  21.     }
  22.     mdwrk_destroy (&session);
  23.     return 0;
  24. }

几点说明：

  • API是单线程的，所以说worker不会再后台发送心跳，而这也是我们所期望的：如果worker应用程序停止了，心跳就会跟着中止，代理便会停止向该worker发送新的请求。

  • wroker API没有做回退算法的设置，因为这里不值得使用这一复杂的机制。

  • API没有提供任何报错机制，如果出现问题，它会直接报断言（或异常，依语言而定）。这一做法对实验性的编程是有用的，这样可以立刻看到执行结果。但在真实编程环境中，API应该足够健壮，合适地处理非法消息。

也许你会问，worker API为什么要关闭它的套接字并新开一个呢？特别是ZMQ是有重连机制的，能够在节点归来后进行重连。我们可以回顾一下简单海盗模式中的worker，以及偏执海盗模式中的worker来加以理解。ZMQ确实会进行自动重连，但如果代理死亡并重连，worker并不会重新进行注册。这个问题有两种解决方案：一是我们这里用到的较为简便的方案，即当worker判断代理已经死亡时，关闭它的套接字并重头来过；另一个方案是当代理收到未知worker的心跳时要求该worker对其提供的服务类型进行注册，这样一来就需要在协议中说明这一规则。

下面让我们设计管家模式的代理，它的核心代码是一组队列，每种服务对应一个队列。我们会在worker出现时创建相应的队列（worker消失时应该销毁对应的队列，不过我们这里暂时不考虑）。额外的，我们会为每种服务维护一个worker的队列。

为了让C语言代码更为易读易写，我使用了ZFL项目提供的哈希和链表容器，并命名为zhash和zlist。如果使用现代语言编写，那自然可以使用其内置的容器。

mdbroker: Majordomo broker in C

  1. //
  2. //  管家模式协议 - 代理
  3. //  协议 http://rfc.zeromq.org/spec:7 和 spec:8 的最简实现
  4. //
  5. #include "czmq.h"
  6. #include "mdp.h"
  8. //  一般我们会从配置文件中获取以下值
  10. #define HEARTBEAT_LIVENESS  3       //  合理值：3-5
  11. #define HEARTBEAT_INTERVAL  2500    //  单位：毫秒
  12. #define HEARTBEAT_EXPIRY    HEARTBEAT_INTERVAL * HEARTBEAT_LIVENESS
  14. //  定义一个代理
  15. typedef struct {
  16.     zctx_t *ctx;                //  上下文
  17.     void *socket;               //  用于连接client和worker的套接字
  18.     int verbose;                //  使用标准输出打印活动信息
  19.     char *endpoint;             //  代理绑定到的端点
  20.     zhash_t *services;          //  已知服务的哈希表
  21.     zhash_t *workers;           //  已知worker的哈希表
  22.     zlist_t *waiting;           //  正在等待的worker队列
  23.     uint64_t heartbeat_at;      //  发送心跳的时间
  24. } broker_t;
  26. //  定义一个服务
  27. typedef struct {
  28.     char *name;                 //  服务名称
  29.     zlist_t *requests;          //  客户端请求队列
  30.     zlist_t *waiting;           //  正在等待的worker队列
  31.     size_t workers;             //  可用worker数
  32. } service_t;
  34. //  定义一个worker，状态为空闲或占用
  35. typedef struct {
  36.     char *identity;             //  worker的标识
  37.     zframe_t *address;          //  地址帧
  38.     service_t *service;         //  所属服务
  39.     int64_t expiry;             //  过期时间，从未收到心跳起计时
  40. } worker_t;
  43. //  ---------------------------------------------------------------------
  44. //  代理使用的函数
  45. static broker_t *
  46.     s_broker_new (int verbose);
  47. static void
  48.     s_broker_destroy (broker_t **self_p);
  49. static void
  50.     s_broker_bind (broker_t *self, char *endpoint);
  51. static void
  52.     s_broker_purge_workers (broker_t *self);
  54. //  服务使用的函数
  55. static service_t *
  56.     s_service_require (broker_t *self, zframe_t *service_frame);
  57. static void
  58.     s_service_destroy (void *argument);
  59. static void
  60.     s_service_dispatch (broker_t *self, service_t *service, zmsg_t *msg);
  61. static void
  62.     s_service_internal (broker_t *self, zframe_t *service_frame, zmsg_t *msg);
  64. //  worker使用的函数
  65. static worker_t *
  66.     s_worker_require (broker_t *self, zframe_t *address);
  67. static void
  68.     s_worker_delete (broker_t *self, worker_t *worker, int disconnect);
  69. static void
  70.     s_worker_destroy (void *argument);
  71. static void
  72.     s_worker_process (broker_t *self, zframe_t *sender, zmsg_t *msg);
  73. static void
  74.     s_worker_send (broker_t *self, worker_t *worker, char *command,
  75.                    char *option, zmsg_t *msg);
  76. static void
  77.     s_worker_waiting (broker_t *self, worker_t *worker);
  79. //  客户端使用的函数
  80. static void
  81.     s_client_process (broker_t *self, zframe_t *sender, zmsg_t *msg);
  84. //  ---------------------------------------------------------------------
  85. //  主程序
  87. int main (int argc, char *argv [])
  88. {
  89.     int verbose = (argc > 1 && streq (argv [1], "-v"));
  91.     broker_t *self = s_broker_new (verbose);
  92.     s_broker_bind (self, "tcp://*:5555");
  94.     //  接受并处理消息，直至程序被中止
  95.     while (TRUE) {
  96.         zmq_pollitem_t items [] = {
  97.             { self->socket,  0, ZMQ_POLLIN, 0 } };
  98.         int rc = zmq_poll (items, 1, HEARTBEAT_INTERVAL * ZMQ_POLL_MSEC);
  99.         if (rc == -1)
  100.             break;              //  中断
  102.         //  Process next input message, if any
  103.         if (items [0].revents & ZMQ_POLLIN) {
  104.             zmsg_t *msg = zmsg_recv (self->socket);
  105.             if (!msg)
  106.                 break;          //  中断
  107.             if (self->verbose) {
  108.                 zclock_log ("I: 收到消息:");
  109.                 zmsg_dump (msg);
  110.             }
  111.             zframe_t *sender = zmsg_pop (msg);
  112.             zframe_t *empty  = zmsg_pop (msg);
  113.             zframe_t *header = zmsg_pop (msg);
  115.             if (zframe_streq (header, MDPC_CLIENT))
  116.                 s_client_process (self, sender, msg);
  117.             else
  118.             if (zframe_streq (header, MDPW_WORKER))
  119.                 s_worker_process (self, sender, msg);
  120.             else {
  121.                 zclock_log ("E: 非法消息:");
  122.                 zmsg_dump (msg);
  123.                 zmsg_destroy (&msg);
  124.             }
  125.             zframe_destroy (&sender);
  126.             zframe_destroy (&empty);
  127.             zframe_destroy (&header);
  128.         }
  129.         //  断开并删除过期的worker
  130.         //  适时地发送心跳给worker
  131.         if (zclock_time () > self->heartbeat_at) {
  132.             s_broker_purge_workers (self);
  133.             worker_t *worker = (worker_t *) zlist_first (self->waiting);
  134.             while (worker) {
  135.                 s_worker_send (self, worker, MDPW_HEARTBEAT, NULL, NULL);
  136.                 worker = (worker_t *) zlist_next (self->waiting);
  137.             }
  138.             self->heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;
  139.         }
  140.     }
  141.     if (zctx_interrupted)
  142.         printf ("W: 收到中断消息，关闭中...\n");
  144.     s_broker_destroy (&self);
  145.     return 0;
  146. }
  149. //  ---------------------------------------------------------------------
  150. //  代理对象的构造函数
  152. static broker_t *
  153. s_broker_new (int verbose)
  154. {
  155.     broker_t *self = (broker_t *) zmalloc (sizeof (broker_t));
  157.     //  初始化代理状态
  158.     self->ctx = zctx_new ();
  159.     self->socket = zsocket_new (self->ctx, ZMQ_ROUTER);
  160.     self->verbose = verbose;
  161.     self->services = zhash_new ();
  162.     self->workers = zhash_new ();
  163.     self->waiting = zlist_new ();
  164.     self->heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;
  165.     return self;
  166. }
  168. //  ---------------------------------------------------------------------
  169. //  代理对象的析构函数
  171. static void
  172. s_broker_destroy (broker_t **self_p)
  173. {
  174.     assert (self_p);
  175.     if (*self_p) {
  176.         broker_t *self = *self_p;
  177.         zctx_destroy (&self->ctx);
  178.         zhash_destroy (&self->services);
  179.         zhash_destroy (&self->workers);
  180.         zlist_destroy (&self->waiting);
  181.         free (self);
  182.         *self_p = NULL;
  183.     }
  184. }
  186. //  ---------------------------------------------------------------------
  187. //  将代理套接字绑定至端点，可以重复调用该函数
  188. //  我们使用一个套接字来同时处理client和worker
  190. void
  191. s_broker_bind (broker_t *self, char *endpoint)
  192. {
  193.     zsocket_bind (self->socket, endpoint);
  194.     zclock_log ("I: MDP broker/0.1.1 is active at %s", endpoint);
  195. }
  197. //  ---------------------------------------------------------------------
  198. //  删除空闲状态中过期的worker
  200. static void
  201. s_broker_purge_workers (broker_t *self)
  202. {
  203.     worker_t *worker = (worker_t *) zlist_first (self->waiting);
  204.     while (worker) {
  205.         if (zclock_time () < worker->expiry)
  206.             continue;              //  该worker未过期，停止搜索
  207.         if (self->verbose)
  208.             zclock_log ("I: 正在删除过期的worker: %s",
  209.                 worker->identity);
  211.         s_worker_delete (self, worker, 0);
  212.         worker = (worker_t *) zlist_first (self->waiting);
  213.     }
  214. }
  216. //  ---------------------------------------------------------------------
  217. //  定位或创建新的服务项
  219. static service_t *
  220. s_service_require (broker_t *self, zframe_t *service_frame)
  221. {
  222.     assert (service_frame);
  223.     char *name = zframe_strdup (service_frame);
  225.     service_t *service =
  226.         (service_t *) zhash_lookup (self->services, name);
  227.     if (service == NULL) {
  228.         service = (service_t *) zmalloc (sizeof (service_t));
  229.         service->name = name;
  230.         service->requests = zlist_new ();
  231.         service->waiting = zlist_new ();
  232.         zhash_insert (self->services, name, service);
  233.         zhash_freefn (self->services, name, s_service_destroy);
  234.         if (self->verbose)
  235.             zclock_log ("I: 收到消息:");
  236.     }
  237.     else
  238.         free (name);
  240.     return service;
  241. }
  243. //  ---------------------------------------------------------------------
  244. //  当服务从broker->services中移除时销毁该服务对象
  246. static void
  247. s_service_destroy (void *argument)
  248. {
  249.     service_t *service = (service_t *) argument;
  250.     //  销毁请求队列中的所有项目
  251.     while (zlist_size (service->requests)) {
  252.         zmsg_t *msg = zlist_pop (service->requests);
  253.         zmsg_destroy (&msg);
  254.     }
  255.     zlist_destroy (&service->requests);
  256.     zlist_destroy (&service->waiting);
  257.     free (service->name);
  258.     free (service);
  259. }
  261. //  ---------------------------------------------------------------------
  262. //  可能时，分发请求给等待中的worker
  264. static void
  265. s_service_dispatch (broker_t *self, service_t *service, zmsg_t *msg)
  266. {
  267.     assert (service);
  268.     if (msg)                    //  将消息加入队列
  269.         zlist_append (service->requests, msg);
  271.     s_broker_purge_workers (self);
  272.     while (zlist_size (service->waiting)
  273.         && zlist_size (service->requests))
  274.     {
  275.         worker_t *worker = zlist_pop (service->waiting);
  276.         zlist_remove (self->waiting, worker);
  277.         zmsg_t *msg = zlist_pop (service->requests);
  278.         s_worker_send (self, worker, MDPW_REQUEST, NULL, msg);
  279.         zmsg_destroy (&msg);
  280.     }
  281. }
  283. //  ---------------------------------------------------------------------
  284. //  使用8/MMI协定处理内部服务
  286. static void
  287. s_service_internal (broker_t *self, zframe_t *service_frame, zmsg_t *msg)
  288. {
  289.     char *return_code;
  290.     if (zframe_streq (service_frame, "mmi.service")) {
  291.         char *name = zframe_strdup (zmsg_last (msg));
  292.         service_t *service =
  293.             (service_t *) zhash_lookup (self->services, name);
  294.         return_code = service && service->workers? "200": "404";
  295.         free (name);
  296.     }
  297.     else
  298.         return_code = "501";
  300.     zframe_reset (zmsg_last (msg), return_code, strlen (return_code));
  302.     //  移除并保存返回给client的信封，插入协议头信息和服务名称，并重新包装信封
  303.     zframe_t *client = zmsg_unwrap (msg);
  304.     zmsg_push (msg, zframe_dup (service_frame));
  305.     zmsg_pushstr (msg, MDPC_CLIENT);
  306.     zmsg_wrap (msg, client);
  307.     zmsg_send (&msg, self->socket);
  308. }
  310. //  ---------------------------------------------------------------------
  311. //  按需创建worker
  313. static worker_t *
  314. s_worker_require (broker_t *self, zframe_t *address)
  315. {
  316.     assert (address);
  318.     //  self->workers使用wroker的标识为键
  319.     char *identity = zframe_strhex (address);
  320.     worker_t *worker =
  321.         (worker_t *) zhash_lookup (self->workers, identity);
  323.     if (worker == NULL) {
  324.         worker = (worker_t *) zmalloc (sizeof (worker_t));
  325.         worker->identity = identity;
  326.         worker->address = zframe_dup (address);
  327.         zhash_insert (self->workers, identity, worker);
  328.         zhash_freefn (self->workers, identity, s_worker_destroy);
  329.         if (self->verbose)
  330.             zclock_log ("I: 正在注册新的worker: %s", identity);
  331.     }
  332.     else
  333.         free (identity);
  334.     return worker;
  335. }
  337. //  ---------------------------------------------------------------------
  338. //  从所有数据结构中删除wroker，并销毁worker对象
  340. static void
  341. s_worker_delete (broker_t *self, worker_t *worker, int disconnect)
  342. {
  343.     assert (worker);
  344.     if (disconnect)
  345.         s_worker_send (self, worker, MDPW_DISCONNECT, NULL, NULL);
  347.     if (worker->service) {
  348.         zlist_remove (worker->service->waiting, worker);
  349.         worker->service->workers--;
  350.     }
  351.     zlist_remove (self->waiting, worker);
  352.     //  以下方法间接调用了s_worker_destroy()方法
  353.     zhash_delete (self->workers, worker->identity);
  354. }
  356. //  ---------------------------------------------------------------------
  357. //  当worker从broker->workers中移除时，销毁worker对象
  359. static void
  360. s_worker_destroy (void *argument)
  361. {
  362.     worker_t *worker = (worker_t *) argument;
  363.     zframe_destroy (&worker->address);
  364.     free (worker->identity);
  365.     free (worker);
  366. }
  368. //  ---------------------------------------------------------------------
  369. //  处理worker发送来的消息
  371. static void
  372. s_worker_process (broker_t *self, zframe_t *sender, zmsg_t *msg)
  373. {
  374.     assert (zmsg_size (msg) >= 1);     //  消息中至少包含命令帧
  376.     zframe_t *command = zmsg_pop (msg);
  377.     char *identity = zframe_strhex (sender);
  378.     int worker_ready = (zhash_lookup (self->workers, identity) != NULL);
  379.     free (identity);
  380.     worker_t *worker = s_worker_require (self, sender);
  382.     if (zframe_streq (command, MDPW_READY)) {
  383.         //  若worker队列中已有该worker，但仍收到了它的“已就绪”消息，则删除这个worker。
  384.         if (worker_ready)
  385.             s_worker_delete (self, worker, 1);
  386.         else
  387.         if (zframe_size (sender) >= 4  //  服务名称为保留的服务
  388.         &&  memcmp (zframe_data (sender), "mmi.", 4) == 0)
  389.             s_worker_delete (self, worker, 1);
  390.         else {
  391.             //  将worker对应到服务，并置为空闲状态
  392.             zframe_t *service_frame = zmsg_pop (msg);
  393.             worker->service = s_service_require (self, service_frame);
  394.             worker->service->workers++;
  395.             s_worker_waiting (self, worker);
  396.             zframe_destroy (&service_frame);
  397.         }
  398.     }
  399.     else
  400.     if (zframe_streq (command, MDPW_REPLY)) {
  401.         if (worker_ready) {
  402.             //  移除并保存返回给client的信封，插入协议头信息和服务名称，并重新包装信封
  403.             zframe_t *client = zmsg_unwrap (msg);
  404.             zmsg_pushstr (msg, worker->service->name);
  405.             zmsg_pushstr (msg, MDPC_CLIENT);
  406.             zmsg_wrap (msg, client);
  407.             zmsg_send (&msg, self->socket);
  408.             s_worker_waiting (self, worker);
  409.         }
  410.         else
  411.             s_worker_delete (self, worker, 1);
  412.     }
  413.     else
  414.     if (zframe_streq (command, MDPW_HEARTBEAT)) {
  415.         if (worker_ready)
  416.             worker->expiry = zclock_time () + HEARTBEAT_EXPIRY;
  417.         else
  418.             s_worker_delete (self, worker, 1);
  419.     }
  420.     else
  421.     if (zframe_streq (command, MDPW_DISCONNECT))
  422.         s_worker_delete (self, worker, 0);
  423.     else {
  424.         zclock_log ("E: 非法消息");
  425.         zmsg_dump (msg);
  426.     }
  427.     free (command);
  428.     zmsg_destroy (&msg);
  429. }
  431. //  ---------------------------------------------------------------------
  432. //  发送消息给worker
  433. //  如果指针指向了一条消息，发送它，但不销毁它，因为这是调用者的工作
  435. static void
  436. s_worker_send (broker_t *self, worker_t *worker, char *command,
  437.                char *option, zmsg_t *msg)
  438. {
  439.     msg = msg? zmsg_dup (msg): zmsg_new ();
  441.     //  将协议信封压入消息顶部
  442.     if (option)
  443.         zmsg_pushstr (msg, option);
  444.     zmsg_pushstr (msg, command);
  445.     zmsg_pushstr (msg, MDPW_WORKER);
  447.     //  在消息顶部插入路由帧
  448.     zmsg_wrap (msg, zframe_dup (worker->address));
  450.     if (self->verbose) {
  451.         zclock_log ("I: 正在发送消息给worker %s",
  452.             mdps_commands [(int) *command]);
  453.         zmsg_dump (msg);
  454.     }
  455.     zmsg_send (&msg, self->socket);
  456. }
  458. //  ---------------------------------------------------------------------
  459. //  正在等待的worker
  461. static void
  462. s_worker_waiting (broker_t *self, worker_t *worker)
  463. {
  464.     //  将worker加入代理和服务的等待队列
  465.     zlist_append (self->waiting, worker);
  466.     zlist_append (worker->service->waiting, worker);
  467.     worker->expiry = zclock_time () + HEARTBEAT_EXPIRY;
  468.     s_service_dispatch (self, worker->service, NULL);
  469. }
  471. //  ---------------------------------------------------------------------
  472. //  处理client发来的请求
  474. static void
  475. s_client_process (broker_t *self, zframe_t *sender, zmsg_t *msg)
  476. {
  477.     assert (zmsg_size (msg) >= 2);     //  服务名称 + 请求内容
  479.     zframe_t *service_frame = zmsg_pop (msg);
  480.     service_t *service = s_service_require (self, service_frame);
  482.     //  为应答内容设置请求方的地址
  483.     zmsg_wrap (msg, zframe_dup (sender));
  484.     if (zframe_size (service_frame) >= 4
  485.     &&  memcmp (zframe_data (service_frame), "mmi.", 4) == 0)
  486.         s_service_internal (self, service_frame, msg);
  487.     else
  488.         s_service_dispatch (self, service, msg);
  489.     zframe_destroy (&service_frame);
  490. }

这个例子应该是我们见过最复杂的一个示例了，大约有500行代码。编写这段代码并让其变的健壮，大约花费了两天的时间。但是，这也仅仅是一个完整的面向服务代理的一部分。

几点说明：

  • 管家模式协议要求我们在一个套接字中同时处理client和worker，这一点对部署和管理代理很有益处：它只会在一个ZMQ端点上收发请求，而不是两个。

  • 代理很好地实现了MDP/0.1协议中规范的内容，包括当代理发送非法命令和心跳时断开的机制。

  • 可以将这段代码扩充为多线程，每个线程管理一个套接字、一组client和worker。这种做法在大型架构的拆分中显得很有趣。C语言代码已经是这样的格式了，因此很容易实现。

  • 还可以将这段代码扩充为主备模式、双在线模式，进一步提高可靠性。因为从本质上来说，代理是无状态的，只是保存了服务的存在与否，因此client和worker可以自行选择除此之外的代理来进行通信。

  • 示例代码中心跳的间隔为5秒，主要是为了减少调试时的输出。现实中的值应该设得低一些，但是，重试的过程应该设置得稍长一些，让服务有足够的时间启动，如10秒钟。

异步管家模式

上文那种实现管家模式的方法比较简单，client还是简单海盗模式中的，仅仅是用API重写了一下。我在测试机上运行了程序，处理10万条请求大约需要14秒的时间，这和代码也有一些关系，因为复制消息帧的时间浪费了CPU处理时间。但真正的问题在于，我们总是逐个循环进行处理（round-trip），即发送-接收-发送-接收……ZMQ内部禁用了TCP发包优化算法（Nagle’s algorithm），但逐个处理循环还是比较浪费。

理论归理论，还是需要由实践来检验。我们用一个简单的测试程序来看看逐个处理循环是否真的耗时。这个测试程序会发送一组消息，第一次它发一条收一条，第二次则一起发送再一起接收。两次结果应该是一样的，但速度截然不同。

tripping: Round-trip demonstrator in C

  1. //
  2. //  Round-trip 模拟
  3. //
  4. //  本示例程序使用多线程的方式启动client、worker、以及代理，
  5. //  当client处理完毕时会发送信号给主程序。
  6. //
  7. #include "czmq.h"
  9. static void
  10. client_task (void *args, zctx_t *ctx, void *pipe)
  11. {
  12.     void *client = zsocket_new (ctx, ZMQ_DEALER);
  13.     zmq_setsockopt (client, ZMQ_IDENTITY, "C", 1);
  14.     zsocket_connect (client, "tcp://localhost:5555");
  16.     printf ("开始测试...\n");
  17.     zclock_sleep (100);
  19.     int requests;
  20.     int64_t start;
  22.     printf ("同步 round-trip 测试...\n");
  23.     start = zclock_time ();
  24.     for (requests = 0; requests < 10000; requests++) {
  25.         zstr_send (client, "hello");
  26.         char *reply = zstr_recv (client);
  27.         free (reply);
  28.     }
  29.     printf (" %d 次/秒\n",
  30.         (1000 * 10000) / (int) (zclock_time () - start));
  32.     printf ("异步 round-trip 测试...\n");
  33.     start = zclock_time ();
  34.     for (requests = 0; requests < 100000; requests++)
  35.         zstr_send (client, "hello");
  36.     for (requests = 0; requests < 100000; requests++) {
  37.         char *reply = zstr_recv (client);
  38.         free (reply);
  39.     }
  40.     printf (" %d 次/秒\n",
  41.         (1000 * 100000) / (int) (zclock_time () - start));
  43.     zstr_send (pipe, "完成");
  44. }
  46. static void *
  47. worker_task (void *args)
  48. {
  49.     zctx_t *ctx = zctx_new ();
  50.     void *worker = zsocket_new (ctx, ZMQ_DEALER);
  51.     zmq_setsockopt (worker, ZMQ_IDENTITY, "W", 1);
  52.     zsocket_connect (worker, "tcp://localhost:5556");
  54.     while (1) {
  55.         zmsg_t *msg = zmsg_recv (worker);
  56.         zmsg_send (&msg, worker);
  57.     }
  58.     zctx_destroy (&ctx);
  59.     return NULL;
  60. }
  62. static void *
  63. broker_task (void *args)
  64. {
  65.     //  准备上下文和套接字
  66.     zctx_t *ctx = zctx_new ();
  67.     void *frontend = zsocket_new (ctx, ZMQ_ROUTER);
  68.     void *backend = zsocket_new (ctx, ZMQ_ROUTER);
  69.     zsocket_bind (frontend, "tcp://*:5555");
  70.     zsocket_bind (backend,  "tcp://*:5556");
  72.     //  初始化轮询对象
  73.     zmq_pollitem_t items [] = {
  74.         { frontend, 0, ZMQ_POLLIN, 0 },
  75.         { backend,  0, ZMQ_POLLIN, 0 }
  76.     };
  77.     while (1) {
  78.         int rc = zmq_poll (items, 2, -1);
  79.         if (rc == -1)
  80.             break;              //  中断
  81.         if (items [0].revents & ZMQ_POLLIN) {
  82.             zmsg_t *msg = zmsg_recv (frontend);
  83.             zframe_t *address = zmsg_pop (msg);
  84.             zframe_destroy (&address);
  85.             zmsg_pushstr (msg, "W");
  86.             zmsg_send (&msg, backend);
  87.         }
  88.         if (items [1].revents & ZMQ_POLLIN) {
  89.             zmsg_t *msg = zmsg_recv (backend);
  90.             zframe_t *address = zmsg_pop (msg);
  91.             zframe_destroy (&address);
  92.             zmsg_pushstr (msg, "C");
  93.             zmsg_send (&msg, frontend);
  94.         }
  95.     }
  96.     zctx_destroy (&ctx);
  97.     return NULL;
  98. }
  100. int main (void)
  101. {
  102.     //  创建线程
  103.     zctx_t *ctx = zctx_new ();
  104.     void *client = zthread_fork (ctx, client_task, NULL);
  105.     zthread_new (ctx, worker_task, NULL);
  106.     zthread_new (ctx, broker_task, NULL);
  108.     //  等待client端管道的信号
  109.     char *signal = zstr_recv (client);
  110.     free (signal);
  112.     zctx_destroy (&ctx);
  113.     return 0;
  114. }

在我的开发环境中运行结果如下：

  1. Setting up test...
  2. Synchronous round-trip test...
  3.  9057 calls/second
  4. Asynchronous round-trip test...
  5.  173010 calls/second

需要注意的是client在运行开始会暂停一段时间，这是因为在向ROUTER套接字发送消息时，若指定标识的套接字没有连接，那么ROUTER会直接丢弃该消息。这个示例中我们没有使用LRU算法，所以当worker连接速度稍慢时就有可能丢失数据，影响测试结果。

我们可以看到，逐个处理循环比异步处理要慢将近20倍，让我们把它应用到管家模式中去。

首先，让我们修改client的API，添加独立的发送和接收方法：

  1. mdcli_t *mdcli_new     (char *broker);
  2. void     mdcli_destroy (mdcli_t **self_p);
  3. int      mdcli_send    (mdcli_t *self, char *service, zmsg_t **request_p);
  4. zmsg_t  *mdcli_recv    (mdcli_t *self);

然后花很短的时间就能将同步的client API改造成异步的API：

mdcliapi2: Majordomo asynchronous client API in C

  1. /*  =====================================================================
  2.     mdcliapi2.c
  4.     Majordomo Protocol Client API (async version)
  5.     Implements the MDP/Worker spec at http://rfc.zeromq.org/spec:7.
  7.     ---------------------------------------------------------------------
  8.     Copyright (c) 1991-2011 iMatix Corporation <www.imatix.com>
  9.     Copyright other contributors as noted in the AUTHORS file.
  11.     This file is part of the ZeroMQ Guide: http://zguide.zeromq.org
  13.     This is free software; you can redistribute it and/or modify it under
  14.     the terms of the GNU Lesser General Public License as published by
  15.     the Free Software Foundation; either version 3 of the License, or (at
  16.     your option) any later version.
  18.     This software is distributed in the hope that it will be useful, but
  19.     WITHOUT ANY WARRANTY; without even the implied warranty of
  20.     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  21.     Lesser General Public License for more details.
  23.     You should have received a copy of the GNU Lesser General Public
  24.     License along with this program. If not, see
  25.     <http://www.gnu.org/licenses/>.
  26.     =====================================================================
  27. */
  29. #include "mdcliapi2.h"
  31. //  类结构
  32. //  使用成员函数访问属性
  34. struct _mdcli_t {
  35.     zctx_t *ctx;                //  上下文
  36.     char *broker;
  37.     void *client;               //  连接至代理的套接字
  38.     int verbose;                //  在标准输出打印运行状态
  39.     int timeout;                //  请求超时时间
  40. };
  43. //  ---------------------------------------------------------------------
  44. //  连接或重连代理
  46. void s_mdcli_connect_to_broker (mdcli_t *self)
  47. {
  48.     if (self->client)
  49.         zsocket_destroy (self->ctx, self->client);
  50.     self->client = zsocket_new (self->ctx, ZMQ_DEALER);
  51.     zmq_connect (self->client, self->broker);
  52.     if (self->verbose)
  53.         zclock_log ("I: 正在连接代理 %s...", self->broker);
  54. }
  57. //  ---------------------------------------------------------------------
  58. //  构造函数
  60. mdcli_t *
  61. mdcli_new (char *broker, int verbose)
  62. {
  63.     assert (broker);
  65.     mdcli_t *self = (mdcli_t *) zmalloc (sizeof (mdcli_t));
  66.     self->ctx = zctx_new ();
  67.     self->broker = strdup (broker);
  68.     self->verbose = verbose;
  69.     self->timeout = 2500;           //  毫秒
  71.     s_mdcli_connect_to_broker (self);
  72.     return self;
  73. }
  76. //  ---------------------------------------------------------------------
  77. //  析构函数
  79. void
  80. mdcli_destroy (mdcli_t **self_p)
  81. {
  82.     assert (self_p);
  83.     if (*self_p) {
  84.         mdcli_t *self = *self_p;
  85.         zctx_destroy (&self->ctx);
  86.         free (self->broker);
  87.         free (self);
  88.         *self_p = NULL;
  89.     }
  90. }
  93. //  ---------------------------------------------------------------------
  94. //  设置请求超时时间
  96. void
  97. mdcli_set_timeout (mdcli_t *self, int timeout)
  98. {
  99.     assert (self);
  100.     self->timeout = timeout;
  101. }
  104. //  ---------------------------------------------------------------------
  105. //  发送请求给代理
  106. //  取得请求消息的所有权，发送后销毁
  108. int
  109. mdcli_send (mdcli_t *self, char *service, zmsg_t **request_p)
  110. {
  111.     assert (self);
  112.     assert (request_p);
  113.     zmsg_t *request = *request_p;
  115.     //  在消息顶部加入协议规定的帧
  116.     //  Frame 0: empty (模拟REQ套接字的行为)
  117.     //  Frame 1: "MDPCxy" (6个字节, MDP/Client x.y)
  118.     //  Frame 2: Service name (看打印字符串)
  119.     zmsg_pushstr (request, service);
  120.     zmsg_pushstr (request, MDPC_CLIENT);
  121.     zmsg_pushstr (request, "");
  122.     if (self->verbose) {
  123.         zclock_log ("I: 发送请求给 '%s' 服务:", service);
  124.         zmsg_dump (request);
  125.     }
  126.     zmsg_send (&request, self->client);
  127.     return 0;
  128. }
  131. //  ---------------------------------------------------------------------
  132. //  获取应答消息，若无则返回NULL；
  133. //  该函数不会尝试从代理的崩溃中恢复，
  134. //  因为我们没有记录那些未收到应答的请求，所以也无法重发。
  136. zmsg_t *
  137. mdcli_recv (mdcli_t *self)
  138. {
  139.     assert (self);
  141.     //  轮询套接字以获取应答
  142.     zmq_pollitem_t items [] = { { self->client, 0, ZMQ_POLLIN, 0 } };
  143.     int rc = zmq_poll (items, 1, self->timeout * ZMQ_POLL_MSEC);
  144.     if (rc == -1)
  145.         return NULL;            //  中断
  147.     //  收到应答后进行处理
  148.     if (items [0].revents & ZMQ_POLLIN) {
  149.         zmsg_t *msg = zmsg_recv (self->client);
  150.         if (self->verbose) {
  151.             zclock_log ("I: received reply:");
  152.             zmsg_dump (msg);
  153.         }
  154.         //  不要处理错误，直接报出
  155.         assert (zmsg_size (msg) >= 4);
  157.         zframe_t *empty = zmsg_pop (msg);
  158.         assert (zframe_streq (empty, ""));
  159.         zframe_destroy (&empty);
  161.         zframe_t *header = zmsg_pop (msg);
  162.         assert (zframe_streq (header, MDPC_CLIENT));
  163.         zframe_destroy (&header);
  165.         zframe_t *service = zmsg_pop (msg);
  166.         zframe_destroy (&service);
  168.         return msg;     //  Success
  169.     }
  170.     if (zctx_interrupted)
  171.         printf ("W: 收到中断消息，正在中止client...\n");
  172.     else
  173.     if (self->verbose)
  174.         zclock_log ("W: 严重错误，放弃请求");
  176.     return NULL;
  177. }

下面是对应的测试代码：

mdclient2: Majordomo client application in C

  1. //
  2. //  异步管家模式 - client示例程序
  3. //  使用mdcli API隐藏MDP协议的具体实现
  4. //
  5. //  直接编译源码，而不创建类库
  6. #include "mdcliapi2.c"
  8. int main (int argc, char *argv [])
  9. {
  10.     int verbose = (argc > 1 && streq (argv [1], "-v"));
  11.     mdcli_t *session = mdcli_new ("tcp://localhost:5555", verbose);
  13.     int count;
  14.     for (count = 0; count < 100000; count++) {
  15.         zmsg_t *request = zmsg_new ();
  16.         zmsg_pushstr (request, "Hello world");
  17.         mdcli_send (session, "echo", &request);
  18.     }
  19.     for (count = 0; count < 100000; count++) {
  20.         zmsg_t *reply = mdcli_recv (session);
  21.         if (reply)
  22.             zmsg_destroy (&reply);
  23.         else
  24.             break;              //  使用Ctrl-C中断
  25.     }
  26.     printf ("收到 %d 个应答\n", count);
  27.     mdcli_destroy (&session);
  28.     return 0;
  29. }

代理和worker的代码没有变，因为我们并没有改变MDP协议。经过对client的改造，我们可以明显看到速度的提升。如以下是同步状况下处理10万条请求的时间：

  1. $ time mdclient
  2. 100000 requests/replies processed
  4. real    0m14.088s
  5. user    0m1.310s
  6. sys     0m2.670s

以下是异步请求的情况：

  1. $ time mdclient2
  2. 100000 replies received
  4. real    0m8.730s
  5. user    0m0.920s
  6. sys     0m1.550s

让我们建立10个worker，看看效果如何：

  1. $ time mdclient2
  2. 100000 replies received
  4. real    0m3.863s
  5. user    0m0.730s
  6. sys     0m0.470s

由于worker获得消息需要通过LRU队列机制，所以并不能做到完全的异步。但是，worker越多其效果也会越好。在我的测试机上，当worker的数量达到8个时，速度就不再提升了——四核处理器只能做这么多。但是，我们仍然获得了近四倍的速度提升，而改造过程只有几分钟而已。此外，代理其实还没有进行优化，它仍会复制消息，而没有实现零拷贝。不过，我们已经做到每秒处理2.5万次请求-应答，已经很不错了。

当然，异步的管家模式也并不完美，有一个显著的缺点：它无法从代理的崩溃中恢复。可以看到mdcliapi2的代码中并没有恢复连接的代码，重新连接需要有以下几点作为前提：

  • 每个请求都做了编号，每次应答也含有相应的编号，这就需要修改协议，明确定义；
  • client的API需要保留并跟踪所有已发送、但仍未收到应答的请求；
  • 如果代理发生崩溃，client会重发所有消息。

可以看到，高可靠性往往和复杂度成正比，值得在管家模式中应用这一机制吗？这就要看应用场景了。如果是一个名称查询服务，每次会话会调用一次，那不需要应用这一机制；如果是一个位于前端的网页服务，有数千个客户端相连，那可能就需要了。

服务查询

现在，我们已经有了一个面向服务的代理了，但是我们无法得知代理是否提供了某项特定服务。如果请求失败，那当然就表示该项服务目前不可用，但具体原因是什么呢？所以，如果能够询问代理“echo服务正在运行吗？”，那将会很有用处。最明显的方法是在MDP/Client协议中添加一种命令，客户端可以询问代理某项服务是否可用。但是，MDP/Client最大的优点在于简单，如果添加了服务查询的功能就太过复杂了。

另一种方案是学电子邮件的处理方式，将失败的请求重新返回。但是这同样会增加复杂度，因为我们需要鉴别收到的消息是一个应答还是被退回的请求。

让我们用之前的方式，在MDP的基础上建立新的机制，而不是改变它。服务定位本身也是一项服务，我们还可以提供类似于“禁用某服务”、“提供服务数据”等其他服务。我们需要的是一个能够扩展协议但又不会影响协议本身的机制。

这样就诞生了一个小巧的RFC - MMI（管家接口）的应用层，建立在MDP协议之上：http://rfc.zeromq.org/spec:8 。我们在代理中其实已经加以实现了，不知你是否已经注意到。下面的代码演示了如何使用这项服务查询功能：

mmiecho: Service discovery over Majordomo in C

  1. //
  2. //  MMI echo 服务查询示例程序
  3. //
  5. //  让我们直接编译，不生成类库
  6. #include "mdcliapi.c"
  8. int main (int argc, char *argv [])
  9. {
  10.     int verbose = (argc > 1 && streq (argv [1], "-v"));
  11.     mdcli_t *session = mdcli_new ("tcp://localhost:5555", verbose);
  13.     //  我们需要查询的服务名称
  14.     zmsg_t *request = zmsg_new ();
  15.     zmsg_addstr (request, "echo");
  17.     //  发送给“服务查询”服务的消息
  18.     zmsg_t *reply = mdcli_send (session, "mmi.service", &request);
  20.     if (reply) {
  21.         char *reply_code = zframe_strdup (zmsg_first (reply));
  22.         printf ("查询 echo 服务的结果: %s\n", reply_code);
  23.         free (reply_code);
  24.         zmsg_destroy (&reply);
  25.     }
  26.     else
  27.         printf ("E: 代理无响应，请确认它正在工作\n");
  29.     mdcli_destroy (&session);
  30.     return 0;
  31. }

代理在运行时会检查请求的服务名称，自行处理那些mmi.开头的服务，而不转发给worker。你可以在不开启worker的情况下运行以上代码，可以看到程序是报告200还是404。MMI在示例程序代理中的实现是很简单的，比如，当某个worker消亡时，该服务仍然标记为可用。实践中，代理应该在一定间隔后清除那些没有worker的服务。

幂等服务

幂等是指能够安全地重复执行某项操作。如，看钟是幂等的，但借钱给别人老婆就不是了。有些客户端至服务端的通信是幂等的，但有些则不是。幂等的通信示例有：

  • 无状态的任务分配，即管道模式中服务端是无状态的worker，它的处理结果是根据客户端的请求状态生成的，因此可以重复处理相同的请求；
  • 命名服务中将逻辑地址转化成实际绑定或连接的端点，可以重复查询多次，因此也是幂等的。

非幂等的通信示例有：

  • 日志服务，我们不会希望相同的日志内容被记录多次；
  • 任何会对下游节点有影响的服务，如该服务会向下游节点发送信息，若收到相同的请求，那下游节点收到的信息就是重复的；
  • 当服务修改了某些共享的数据，且没有进行幂等方面的设置。如某项服务对银行账户进行了借操作（debit），这一定是非幂等的。

如果应用程序提供的服务是非幂等的，那就需要考虑它究竟是在哪个阶段崩溃的。如果程序在空闲或处理请求的过程中崩溃，那不会有什么问题。我们可以使用数据库中的事务机制来保证借贷操作是同时发生的。如果应用程序在发送请求的时候崩溃了，那就会有问题，因为对于该程序来说，它已经完成了工作。

如果在返回应答的过程中网络阻塞了，客户端会认为请求发送失败，并进行重发，这样服务端会再一次执行相同的请求。这不是我们想要的结果。

常用的解决方法是在服务端检测并拒绝重复的请求，这就需要：

  • 客户端为每个请求加注唯一的标识，包括客户端标识和消息标识；
  • 服务端在发送应答时使用客户端标识和消息标识作为键，保存应答内容；
  • 当服务端发现收到的请求已在应答哈希表中存在，它会跳过该次请求，直接返回应答内容。

脱机可靠性（巨人模式）

当你意识到管家模式是一种非常可靠的消息代理时，你可能会想要使用磁盘做一下消息中转，从而进一步提升可靠性。这种方式虽然在很多企业级消息系统中应用，但我还是有些反对的，原因有：

  • 我们可以看到，懒惰海盗模式的client可以工作得非常好，能够在多种架构中运行。唯一的问题是它会假设worker是无状态的，且提供的服务是幂等的。但这个问题我们可以通过其他方式解决，而不是添加磁盘。

  • 添加磁盘会带来新的问题，需要额外的管理和维护费用。海盗模式的最大优点就是简单明了，不会崩溃。如果你还是担心硬件会出问题，可以改用点对点的通信模式，这会在本章最后一节讲到。

虽然有以上原因，但还是有一个合理的场景可以用到磁盘中转的——异步脱机网络。海盗模式有一个问题，那就是client发送请求后会一直等待应答。如果client和worker并不是长连接（可以拿电子邮箱做个类比），我们就无法在client和worker之间建立一个无状态的网络，因此需要将这种状态保存起来。

于是我们就有了巨人模式，该模式下会将消息写到磁盘中，确保不会丢失。当我们进行服务查询时，会转向巨人这一层进行。巨人是建立在管家之上的，而不是改写了MDP协议。这样做的好处是我们可以在一个特定的worker中实现这种可靠性，而不用去增加代理的逻辑。

  • 实现更为简单；
    • 代理用一种语言编写，worker使用另一种语言编写；
    • 可以自由升级这种模式。

唯一的缺点是，代理和磁盘之间会有一层额外的联系，不过这也是值得的。

我们有很多方法来实现一种持久化的请求-应答架构，而目标当然是越简单越好。我能想到的最简单的方式是提供一种成为“巨人”的代理服务，它不会影响现有worker的工作，若client想要立即得到应答，它可以和代理进行通信；如果它不是那么着急，那就可以和巨人通信：“嗨，巨人，麻烦帮我处理下这个请求，我去买些菜。”

5

这样一来，巨人就既是worker又是client。client和巨人之间的对话一般是：

  • Client: 请帮我处理这个请求。巨人：好的。
  • Client: 有要给我的应答吗？巨人：有的。（或者没有）
  • Client: OK，你可以释放那个请求了，工作已经完成。巨人：好的。

巨人和代理之间的对话一般是：

  • 巨人：嗨，代理程序，你这里有个叫echo的服务吗？代理：恩，好像有。
  • 巨人：嗨，echo服务，请帮我处理一下这个请求。Echo: 好了，这是应答。
  • 巨人：谢谢！

你可以想象一些发生故障的情形，看看上述模式是否能解决？worker在处理请求的时候崩溃，巨人会不断地重新发送请求；应答在传输过程中丢失了，巨人也会重试；如果请求已经处理，但client没有得到应答，那它会再次询问巨人；如果巨人在处理请求或进行应答的时候崩溃了，客户端会进行重试；只要请求是被保存在磁盘上的，那它就不会丢失。

这个机制中，握手的过程是比较漫长的，但client可以使用异步的管家模式，一次发送多个请求，并一起等待应答。

我们需要一种方法，让client会去请求应答内容。不同的client会访问到相同的服务，且client是来去自由的，有着不同的标识。一个简单、合理、安全的解决方案是：

  • 当巨人收到请求时，它会为每个请求生成唯一的编号（UUID），并将这个编号返回给client；
  • client在请求应答内容时需要提供这个编号。

这样一来client就需要负责将UUID安全地保存起来，不过这就省去了验证的过程。有其他方案吗？我们可以使用持久化的套接字，即显式声明客户端的套接字标识。然而，这会造成管理上的麻烦，而且万一两个client的套接字标识相同，那会引来无穷的麻烦。

在我们开始制定一个新的协议之前，我们先思考一下client如何和巨人通信。一种方案是提供一种服务，配合三个不同的命令；另一种方案则更为简单，提供三种独立的服务：

  • titanic.request - 保存一个请求，并返回UUID
  • titanic.reply - 根据UUID获取应答内容
  • titanic.close - 确认某个请求已被正确地处理

我们需要创建一个多线程的worker，正如我们之前用ZMQ进行多线程编程一样，很简单。但是，在我们开始编写代码之前，先讲巨人模式的一些定义写下来：http://rfc.zeromq.org/spec:9 。我们称之为“巨人服务协议”，或TSP。

使用TSP协议自然会让client多出额外的工作，下面是一个简单但足够健壮的client：

ticlient: Titanic client example in C

  1. //
  2. //  巨人模式client示例
  3. //  实现 http://rfc.zeromq.org/spec:9 协议中的client端
  5. //  让我们直接编译，不创建类库
  6. #include "mdcliapi.c"
  8. //  请求TSP协议下的服务
  9. //  如果成功则返回应答（状态码：200），否则返回NULL
  10. //
  11. static zmsg_t *
  12. s_service_call (mdcli_t *session, char *service, zmsg_t **request_p)
  13. {
  14.     zmsg_t *reply = mdcli_send (session, service, request_p);
  15.     if (reply) {
  16.         zframe_t *status = zmsg_pop (reply);
  17.         if (zframe_streq (status, "200")) {
  18.             zframe_destroy (&status);
  19.             return reply;
  20.         }
  21.         else
  22.         if (zframe_streq (status, "400")) {
  23.             printf ("E: 客户端发生严重错误，取消请求\n");
  24.             exit (EXIT_FAILURE);
  25.         }
  26.         else
  27.         if (zframe_streq (status, "500")) {
  28.             printf ("E: 服务端发生严重错误，取消请求\n");
  29.             exit (EXIT_FAILURE);
  30.         }
  31.     }
  32.     else
  33.         exit (EXIT_SUCCESS);    //  中断或发生错误
  35.     zmsg_destroy (&reply);
  36.     return NULL;        //  请求不成功，但不返回失败原因
  37. }
  39. int main (int argc, char *argv [])
  40. {
  41.     int verbose = (argc > 1 && streq (argv [1], "-v"));
  42.     mdcli_t *session = mdcli_new ("tcp://localhost:5555", verbose);
  44.     //  1. 发送echo服务的请求给巨人
  45.     zmsg_t *request = zmsg_new ();
  46.     zmsg_addstr (request, "echo");
  47.     zmsg_addstr (request, "Hello world");
  48.     zmsg_t *reply = s_service_call (
  49.         session, "titanic.request", &request);
  51.     zframe_t *uuid = NULL;
  52.     if (reply) {
  53.         uuid = zmsg_pop (reply);
  54.         zmsg_destroy (&reply);
  55.         zframe_print (uuid, "I: request UUID ");
  56.     }
  58.     //  2. 等待应答
  59.     while (!zctx_interrupted) {
  60.         zclock_sleep (100);
  61.         request = zmsg_new ();
  62.         zmsg_add (request, zframe_dup (uuid));
  63.         zmsg_t *reply = s_service_call (
  64.             session, "titanic.reply", &request);
  66.         if (reply) {
  67.             char *reply_string = zframe_strdup (zmsg_last (reply));
  68.             printf ("Reply: %s\n", reply_string);
  69.             free (reply_string);
  70.             zmsg_destroy (&reply);
  72.             //  3. 关闭请求
  73.             request = zmsg_new ();
  74.             zmsg_add (request, zframe_dup (uuid));
  75.             reply = s_service_call (session, "titanic.close", &request);
  76.             zmsg_destroy (&reply);
  77.             break;
  78.         }
  79.         else {
  80.             printf ("I: 尚未收到应答，准备稍后重试...\n");
  81.             zclock_sleep (5000);     //  5秒后重试
  82.         }
  83.     }
  84.     zframe_destroy (&uuid);
  85.     mdcli_destroy (&session);
  86.     return 0;
  87. }

当然，上面的代码可以整合到一个框架中，程序员不需要了解其中的细节。如果我有时间的话，我会尝试写一个这样的API的，让应用程序又变回短短的几行。这种理念和MDP中的一致：不要做重复的事。

下面是巨人的实现。这个服务端会使用三个线程来处理三种服务。它使用最原始的持久化方法来保存请求：为每个请求创建一个磁盘文件。虽然简单，但也挺恐怖的。比较复杂的部分是，巨人会维护一个队列来保存这些请求，从而避免重复地扫描目录。

titanic: Titanic broker example in C

  1. //
  2. //  巨人模式 - 服务
  3. //
  4. //  实现 http://rfc.zeromq.org/spec:9 协议的服务端
  6. //  让我们直接编译，不创建类库
  7. #include "mdwrkapi.c"
  8. #include "mdcliapi.c"
  10. #include "zfile.h"
  11. #include <uuid/uuid.h>
  13. //  返回一个可打印的唯一编号（UUID）
  14. //  调用者负责释放UUID字符串的内存
  16. static char *
  17. s_generate_uuid (void)
  18. {
  19.     char hex_char [] = "0123456789ABCDEF";
  20.     char *uuidstr = zmalloc (sizeof (uuid_t) * 2 + 1);
  21.     uuid_t uuid;
  22.     uuid_generate (uuid);
  23.     int byte_nbr;
  24.     for (byte_nbr = 0; byte_nbr < sizeof (uuid_t); byte_nbr++) {
  25.         uuidstr [byte_nbr * 2 + 0] = hex_char [uuid [byte_nbr] >> 4];
  26.         uuidstr [byte_nbr * 2 + 1] = hex_char [uuid [byte_nbr] & 15];
  27.     }
  28.     return uuidstr;
  29. }
  31. //  根据UUID生成用于保存请求内容的文件名，并返回
  33. #define TITANIC_DIR ".titanic"
  35. static char *
  36. s_request_filename (char *uuid) {
  37.     char *filename = malloc (256);
  38.     snprintf (filename, 256, TITANIC_DIR "/%s.req", uuid);
  39.     return filename;
  40. }
  42. //  根据UUID生成用于保存应答内容的文件名，并返回
  44. static char *
  45. s_reply_filename (char *uuid) {
  46.     char *filename = malloc (256);
  47.     snprintf (filename, 256, TITANIC_DIR "/%s.rep", uuid);
  48.     return filename;
  49. }
  52. //  ---------------------------------------------------------------------
  53. //  巨人模式 - 请求服务
  55. static void
  56. titanic_request (void *args, zctx_t *ctx, void *pipe)
  57. {
  58.     mdwrk_t *worker = mdwrk_new (
  59.         "tcp://localhost:5555", "titanic.request", 0);
  60.     zmsg_t *reply = NULL;
  62.     while (TRUE) {
  63.         //  若应答非空则发送，再从代理处获得新的请求
  64.         zmsg_t *request = mdwrk_recv (worker, &reply);
  65.         if (!request)
  66.             break;      //  中断并退出
  68.         //  确保消息目录是存在的
  69.         file_mkdir (TITANIC_DIR);
  71.         //  生成UUID，并将消息保存至磁盘
  72.         char *uuid = s_generate_uuid ();
  73.         char *filename = s_request_filename (uuid);
  74.         FILE *file = fopen (filename, "w");
  75.         assert (file);
  76.         zmsg_save (request, file);
  77.         fclose (file);
  78.         free (filename);
  79.         zmsg_destroy (&request);
  81.         //  将UUID加入队列
  82.         reply = zmsg_new ();
  83.         zmsg_addstr (reply, uuid);
  84.         zmsg_send (&reply, pipe);
  86.         //  将UUID返回给客户端
  87.         //  将由循环顶部的mdwrk_recv()函数完成
  88.         reply = zmsg_new ();
  89.         zmsg_addstr (reply, "200");
  90.         zmsg_addstr (reply, uuid);
  91.         free (uuid);
  92.     }
  93.     mdwrk_destroy (&worker);
  94. }
  97. //  ---------------------------------------------------------------------
  98. //  巨人模式 - 应答服务
  100. static void *
  101. titanic_reply (void *context)
  102. {
  103.     mdwrk_t *worker = mdwrk_new (
  104.         "tcp://localhost:5555", "titanic.reply", 0);
  105.     zmsg_t *reply = NULL;
  107.     while (TRUE) {
  108.         zmsg_t *request = mdwrk_recv (worker, &reply);
  109.         if (!request)
  110.             break;      //  中断并退出
  112.         char *uuid = zmsg_popstr (request);
  113.         char *req_filename = s_request_filename (uuid);
  114.         char *rep_filename = s_reply_filename (uuid);
  115.         if (file_exists (rep_filename)) {
  116.             FILE *file = fopen (rep_filename, "r");
  117.             assert (file);
  118.             reply = zmsg_load (file);
  119.             zmsg_pushstr (reply, "200");
  120.             fclose (file);
  121.         }
  122.         else {
  123.             reply = zmsg_new ();
  124.             if (file_exists (req_filename))
  125.                 zmsg_pushstr (reply, "300"); //挂起
  126.             else
  127.                 zmsg_pushstr (reply, "400"); //未知
  128.         }
  129.         zmsg_destroy (&request);
  130.         free (uuid);
  131.         free (req_filename);
  132.         free (rep_filename);
  133.     }
  134.     mdwrk_destroy (&worker);
  135.     return 0;
  136. }
  139. //  ---------------------------------------------------------------------
  140. //  巨人模式 - 关闭请求
  142. static void *
  143. titanic_close (void *context)
  144. {
  145.     mdwrk_t *worker = mdwrk_new (
  146.         "tcp://localhost:5555", "titanic.close", 0);
  147.     zmsg_t *reply = NULL;
  149.     while (TRUE) {
  150.         zmsg_t *request = mdwrk_recv (worker, &reply);
  151.         if (!request)
  152.             break;      //  中断并退出
  154.         char *uuid = zmsg_popstr (request);
  155.         char *req_filename = s_request_filename (uuid);
  156.         char *rep_filename = s_reply_filename (uuid);
  157.         file_delete (req_filename);
  158.         file_delete (rep_filename);
  159.         free (uuid);
  160.         free (req_filename);
  161.         free (rep_filename);
  163.         zmsg_destroy (&request);
  164.         reply = zmsg_new ();
  165.         zmsg_addstr (reply, "200");
  166.     }
  167.     mdwrk_destroy (&worker);
  168.     return 0;
  169. }
  171. //  处理某个请求，成功则返回1
  173. static int
  174. s_service_success (mdcli_t *client, char *uuid)
  175. {
  176.     //  读取请求内容，第一帧为服务名称
  177.     char *filename = s_request_filename (uuid);
  178.     FILE *file = fopen (filename, "r");
  179.     free (filename);
  181.     //  如果client已经关闭了该请求，则返回1
  182.     if (!file)
  183.         return 1;
  185.     zmsg_t *request = zmsg_load (file);
  186.     fclose (file);
  187.     zframe_t *service = zmsg_pop (request);
  188.     char *service_name = zframe_strdup (service);
  190.     //  使用MMI协议检查服务是否可用
  191.     zmsg_t *mmi_request = zmsg_new ();
  192.     zmsg_add (mmi_request, service);
  193.     zmsg_t *mmi_reply = mdcli_send (client, "mmi.service", &mmi_request);
  194.     int service_ok = (mmi_reply
  195.         && zframe_streq (zmsg_first (mmi_reply), "200"));
  196.     zmsg_destroy (&mmi_reply);
  198.     if (service_ok) {
  199.         zmsg_t *reply = mdcli_send (client, service_name, &request);
  200.         if (reply) {
  201.             filename = s_reply_filename (uuid);
  202.             FILE *file = fopen (filename, "w");
  203.             assert (file);
  204.             zmsg_save (reply, file);
  205.             fclose (file);
  206.             free (filename);
  207.             return 1;
  208.         }
  209.         zmsg_destroy (&reply);
  210.     }
  211.     else
  212.         zmsg_destroy (&request);
  214.     free (service_name);
  215.     return 0;
  216. }
  219. int main (int argc, char *argv [])
  220. {
  221.     int verbose = (argc > 1 && streq (argv [1], "-v"));
  222.     zctx_t *ctx = zctx_new ();
  224.     //  创建MDP客户端会话
  225.     mdcli_t *client = mdcli_new ("tcp://localhost:5555", verbose);
  226.     mdcli_set_timeout (client, 1000);  //  1 秒
  227.     mdcli_set_retries (client, 1);     //  只尝试一次
  229.     void *request_pipe = zthread_fork (ctx, titanic_request, NULL);
  230.     zthread_new (ctx, titanic_reply, NULL);
  231.     zthread_new (ctx, titanic_close, NULL);
  233.     //  主循环
  234.     while (TRUE) {
  235.         //  如果没有活动，我们将每秒循环一次
  236.         zmq_pollitem_t items [] = { { request_pipe, 0, ZMQ_POLLIN, 0 } };
  237.         int rc = zmq_poll (items, 1, 1000 * ZMQ_POLL_MSEC);
  238.         if (rc == -1)
  239.             break;              //  中断
  240.         if (items [0].revents & ZMQ_POLLIN) {
  241.             //  确保消息目录是存在的
  242.             file_mkdir (TITANIC_DIR);
  244.             //  将UUID添加到队列中，使用“-”号标识等待中的请求
  245.             zmsg_t *msg = zmsg_recv (request_pipe);
  246.             if (!msg)
  247.                 break;          //  中断
  248.             FILE *file = fopen (TITANIC_DIR "/queue", "a");
  249.             char *uuid = zmsg_popstr (msg);
  250.             fprintf (file, "-%s\n", uuid);
  251.             fclose (file);
  252.             free (uuid);
  253.             zmsg_destroy (&msg);
  254.         }
  255.         //  分派
  256.         //
  257.         char entry [] = "?.......:.......:.......:.......:";
  258.         FILE *file = fopen (TITANIC_DIR "/queue", "r+");
  259.         while (file && fread (entry, 33, 1, file) == 1) {
  260.             //  处理UUID前缀为“-”的请求
  261.             if (entry [0] == '-') {
  262.                 if (verbose)
  263.                     printf ("I: 开始处理请求 %s\n", entry + 1);
  264.                 if (s_service_success (client, entry + 1)) {
  265.                     //  标记为已处理
  266.                     fseek (file, -33, SEEK_CUR);
  267.                     fwrite ("+", 1, 1, file);
  268.                     fseek (file, 32, SEEK_CUR);
  269.                 }
  270.             }
  271.             //  跳过最后一行
  272.             if (fgetc (file) == '\r')
  273.                 fgetc (file);
  274.             if (zctx_interrupted)
  275.                 break;
  276.         }
  277.         if (file)
  278.             fclose (file);
  279.     }
  280.     mdcli_destroy (&client);
  281.     return 0;
  282. }

测试时，打开mdbroker和titanic，再运行ticlient，然后开启任意个mdworker，就可以看到client获得了应答。

几点说明：

  • 我们使用MMI协议去向代理询问某项服务是否可用，这一点和MDP中的逻辑一致；
  • 我们使用inproc（进程内）协议建立主循环和titanic.request服务间的联系，保存新的请求信息。这样可以避免主循环不断扫描磁盘目录，读取所有请求文件，并按照时间日期排序。

这个示例程序不应关注它的性能（一定会非常糟糕，虽然我没有测试过），而是应该看到它是如何提供一种可靠的通信模式的。你可以测试一下，打开代理、巨人、worker和client，使用-v参数显示跟踪信息，然后随意地开关代理、巨人、或worker（client不能关闭），可以看到所有的请求都能获得应答。

如果你想在真实环境中使用巨人模式，你肯定会问怎样才能让速度快起来。以下是我的做法：

  • 使用一个磁盘文件保存所有数据。操作系统处理大文件的效率要比处理许多小文件来的高。
  • 使用一种循环的机制来组织该磁盘文件的结构，这样新的请求可以被连续地写入这个文件。单个线程在全速写入磁盘时的效率是比较高的。
  • 将索引保存在内存中，可以在启动程序时重建这个索引。这样做可以节省磁盘缓存，让索引安全地保存在磁盘上。你需要用到fsync的机制来保存每一条数据；或者可以等待几毫秒，如果不怕丢失上千条数据的话。
  • 如果条件允许，应选择使用固态硬盘；
  • 提前分配该磁盘文件的空间，或者将每次分配的空间调大一些，这样可以避免磁盘碎片的产生，并保证读写是连续的。

另外，我不建议将消息保存在数据库中，甚至不建议交给那些所谓的高速键值缓存，它们比起一个磁盘文件要来得昂贵。

如果你想让巨人模式变得更为可靠，你可以将请求复制到另一台服务器上，这样就不需要担心主程序遭到核武器袭击了。

如果你想让巨人模式变得更为快速，但可以牺牲一些可靠性，那你可以将请求和应答都保存在内存中。这样做可以让该服务作为脱机网络运行，不过若巨人服务本身崩溃了，我也无能为力。

高可靠对称节点（双子星模式）

概览

双子星模式是一对具有主从机制的高可靠节点。任一时间，某个节点会充当主机，接收所有客户端的请求；另一个则作为一种备机存在。两个节点会互相监控对方，当主机从网络中消失时，备机会替代主机的位置。

双子星模式由Pieter Hintjens和Martin Sustrik设计，应用在iMatix的OpenAMQ服务器中。它的设计理念是：

  • 提供一种简明的高可靠性解决方案；
  • 易于理解和使用；
  • 能够进行可靠的故障切换。

6

假设我们有一组双子星模式的服务器，以下是可能发生的故障：

  1. 主机发生硬件故障（断电、失火等），应用程序发送后立刻使用备机进行连接；
  2. 主机的网络环境发生故障，可能某个路由器被雷击了，立刻使用备机；
  3. 主机上的服务被维护人员误杀，无法自动恢复。

恢复步骤如下：

  1. 维护人员排查主机故障；
  2. 将备机关闭，造成短时间的服务不可用；
  3. 待应用程序都连接到主机后，维护人员重启备机。

恢复过程是人工进行的，惨痛的经验告诉我们自动恢复是很可怕的：

  • 故障的发生会造成10-30秒之间的服务暂停，如果这是一个真正的突发状况，那最好还是让主机暂停服务的好，因为立刻重启服务可能造成另一个10-30秒的暂停，不如让用户停止使用。

  • 当有紧急状况发生时，可以在修复的过程中记录故障发生原因，而不是让系统自动恢复，管理员因此无法用其经验抵御下一次突发状况。

  • 最后，如果自动恢复确实成功了，管理员将无从得知故障的发生原因，因而无法进行分析。

双子星模式的故障恢复过程是：在修复了主机的问题后，将备机做关闭处理，稍后再重新开启：

7

双子星模式的关闭过程有两种：

  1. 先关闭备机，等待一段时间后再关闭主机；
  2. 同时关闭主机和备机，间隔时间不超过几秒。

关闭时，间隔时间要比故障切换时间短，否则会导致应用程序失去连接、重新连接、并再次失去连接，导致用户投诉。

详细要求

双子星模式可以非常简单，但能工作得很出色。事实上，这里的实现方法已经历经三个版本了，之前的版本都过于复杂，想要做太多的事情，因而被我们抛弃。我们需要的只是最基本的功能，能够提供易理解、易开发、高可靠的解决方法就可以了。

以下是该架构的详细需求：

  • 需要用到双子星模式的故障是：系统遭受灾难性的打击，如硬件崩溃、火灾、意外等。对于其他常规的服务器故障，可以用更简单的方法。
  • 故障恢复时间应该在60秒以内，理想情况下应该在10秒以内；
  • 故障恢复(failover)应该是自动完成的，而系统还原(recover)则是由人工完成的。我们希望应用程序能够在发生故障时自动从主机切换到备机，但不希望在问题解决之前自动切换回主机，因为这很有可能让主机再次崩溃。
  • 程序的逻辑应该尽量简单，易于使用，最好能封装在API中；
  • 需要提供一个明确的指示，哪台主机正在提供服务，以避免“精神分裂”的症状，即两台服务器都认为自己是主机；
  • 两台服务器的启动顺序不应该有限制；
  • 启动或关闭主从机时不需要更改客户端的配置，但有可能会中断连接；
  • 管理员需要能够同时监控两台机器；
  • 两台机器之间必须有专用的高速网络连接，必须能使用特定IP进行路由。

我们做如下架假设：

  • 单台备机能够提供足够的保障，不需要再进行其他备份机制；
  • 主从机应该都能够提供完整的服务，承载相同的压力，不需要进行负载均衡；
  • 预算中允许有这样一台长时间闲置的备机。

双子星模式不会用到：

  • 多台备机，或在主从机之间进行负载均衡。该模式中的备机将一直处于空闲状态，只有主机发生问题时才会工作；
  • 处理持久化的消息或事务。我们假设所连接的网络是不可靠的（或不可信的）。
  • 自动搜索网络。双子星模式是手工配置的，他们知道对方的存在，应用程序则知道双子星的存在。
  • 主从机之间状态的同步。所有服务端的状态必须能由应用程序进行重建。

以下是双子星模式中的几个术语：

  • 主机 - 通常情况下作为master的机器；
  • 备机 - 通常情况下作为slave的机器，只有当主机从网络中消失时，备机才会切换成master状态，接收所有的应用程序请求；
  • master - 双子星模式中接收应用程序请求的机器；同一时刻只有一台master；
  • slave - 当master消失时用以顶替的机器。

配置双子星模式的步骤：

  1. 让主机知道备机的位置；
  2. 让备机知道主机的位置；
  3. 调整故障恢复时间，两台机器的配置必须相同。

比较重要的配置是应让两台机器间隔多久检查一次对方的状态，以及多长时间后采取行动。在我们的示例中，故障恢复时间设置为2000毫秒，超过这个时间备机就会代替主机的位置。但若你将主机的服务包裹在一个shell脚本中进行重启，就需要延长这个时间，否则备机可能在主机恢复连接的过程中转换成master。

要让客户端应用程序和双子星模式配合，你需要做的是：

  1. 知道两台服务器的地址；
  2. 尝试连接主机，若失败则连接备机；
  3. 检测失效的连接，一般使用心跳机制；
  4. 尝试重连主机，然后再连接备机，其间的间隔应比服务器故障恢复时间长；
  5. 重建服务器端需要的所有状态数据；
  6. 如果要保证可靠性，应重发故障期间的消息。

这不是件容易的事，所以我们一般会将其封装成一个API，供程序员使用。

双子星模式的主要限制有：

  • 服务端进程不能涉及到一个以上的双子星对称节点；
  • 主机只能有一个备机；
  • 当备机处于slave状态时，它不会处理任何请求；
  • 备机必须能够承受所有的应用程序请求；
  • 故障恢复时间不能在运行时调整；
  • 客户端应用程序需要做一些重连的工作。

防止精神分裂

“精神分裂”症状指的是一个集群中的不同部分同时认为自己是master，从而停止对对方的检测。双子星模式中的算法会降低这种症状的发生几率：主备机在决定自己是否为master时会检测自身是否收到了应用程序的请求，以及对方是否已经从网络中消失。

但在某些情况下，双子星模式也会发生精神分裂。比如说，主备机被配置在两幢大楼里，每幢大楼的局域网中又分布了一些应用程序。这样，当两幢大楼的网络通信被阻断，双子星模式的主备机就会分别在两幢大楼里接受和处理请求。

为了防止精神分裂，我们必须让主备机使用专用的网络进行连接，最简单的方法当然是用一根双绞线将他们相连。

我们不能将双子星部署在两个不同的岛屿上，为各自岛屿的应用程序服务。这种情况下，我们会使用诸如联邦模式的机制进行可靠性设计。

最好但最夸张的做法是，将两台机器之间的连接和应用程序的连接完全隔离开来，甚至是使用不同的网卡，而不仅仅是不同的端口。这样做也是为了日后排查错误时更为明确。

实现双子星模式

闲话少说，下面是双子星模式的服务端代码：

bstarsrv: Binary Star server in C

  1. //
  2. //  双子星模式 - 服务端
  3. //
  4. #include "czmq.h"
  6. //  发送状态信息的间隔时间
  7. //  如果对方在两次心跳过后都没有应答，则视为断开
  8. #define HEARTBEAT 1000          //  In msecs
  10. //  服务器状态枚举
  11. typedef enum {
  12.     STATE_PRIMARY = 1,          //  主机，等待同伴连接
  13.     STATE_BACKUP = 2,           //  备机，等待同伴连接
  14.     STATE_ACTIVE = 3,           //  激活态，处理应用程序请求
  15.     STATE_PASSIVE = 4           //  被动态，不接收请求
  16. } state_t;
  18. //  对话节点事件
  19. typedef enum {
  20.     PEER_PRIMARY = 1,           //  主机
  21.     PEER_BACKUP = 2,            //  备机
  22.     PEER_ACTIVE = 3,            //  激活态
  23.     PEER_PASSIVE = 4,           //  被动态
  24.     CLIENT_REQUEST = 5          //  客户端请求
  25. } event_t;
  27. //  有限状态机
  28. typedef struct {
  29.     state_t state;              //  当前状态
  30.     event_t event;              //  当前事件
  31.     int64_t peer_expiry;        //  判定节点死亡的时限
  32. } bstar_t;
  35. //  执行有限状态机（将事件绑定至状态）；
  36. //  发生异常时返回TRUE。
  38. static Bool
  39. s_state_machine (bstar_t *fsm)
  40. {
  41.     Bool exception = FALSE;
  42.     //  主机等待同伴连接
  43.     //  该状态下接收CLIENT_REQUEST事件
  44.     if (fsm->state == STATE_PRIMARY) {
  45.         if (fsm->event == PEER_BACKUP) {
  46.             printf ("I: 已连接至备机（slave），可以作为master运行。\n");
  47.             fsm->state = STATE_ACTIVE;
  48.         }
  49.         else
  50.         if (fsm->event == PEER_ACTIVE) {
  51.             printf ("I: 已连接至备机（master），可以作为slave运行。\n");
  52.             fsm->state = STATE_PASSIVE;
  53.         }
  54.     }
  55.     else
  56.     //  备机等待同伴连接
  57.     //  该状态下拒绝CLIENT_REQUEST事件
  58.     if (fsm->state == STATE_BACKUP) {
  59.         if (fsm->event == PEER_ACTIVE) {
  60.             printf ("I: 已连接至主机（master），可以作为slave运行。\n");
  61.             fsm->state = STATE_PASSIVE;
  62.         }
  63.         else
  64.         if (fsm->event == CLIENT_REQUEST)
  65.             exception = TRUE;
  66.     }
  67.     else
  68.     //  服务器处于激活态
  69.     //  该状态下接受CLIENT_REQUEST事件
  70.     if (fsm->state == STATE_ACTIVE) {
  71.         if (fsm->event == PEER_ACTIVE) {
  72.             //  若出现两台master，则抛出异常
  73.             printf ("E: 严重错误：双master。正在退出。\n");
  74.             exception = TRUE;
  75.         }
  76.     }
  77.     else
  78.     //  服务器处于被动态
  79.     //  若同伴已死，CLIENT_REQUEST事件将触发故障恢复
  80.     if (fsm->state == STATE_PASSIVE) {
  81.         if (fsm->event == PEER_PRIMARY) {
  82.             //  同伴正在重启 - 转为激活态，同伴将转为被动态。
  83.             printf ("I: 主机（slave）正在重启，可作为master运行。\n");
  84.             fsm->state = STATE_ACTIVE;
  85.         }
  86.         else
  87.         if (fsm->event == PEER_BACKUP) {
  88.             //  同伴正在重启 - 转为激活态，同伴将转为被动态。
  89.             printf ("I: 备机（slave）正在重启，可作为master运行。\n");
  90.             fsm->state = STATE_ACTIVE;
  91.         }
  92.         else
  93.         if (fsm->event == PEER_PASSIVE) {
  94.             //  若出现两台slave，集群将无响应
  95.             printf ("E: 严重错误：双slave。正在退出\n");
  96.             exception = TRUE;
  97.         }
  98.         else
  99.         if (fsm->event == CLIENT_REQUEST) {
  100.             //  若心跳超时，同伴将成为master；
  101.             //  此行为由客户端请求触发。
  102.             assert (fsm->peer_expiry > 0);
  103.             if (zclock_time () >= fsm->peer_expiry) {
  104.                 //  同伴已死，转为激活态。
  105.                 printf ("I: 故障恢复，可作为master运行。\n");
  106.                 fsm->state = STATE_ACTIVE;
  107.             }
  108.             else
  109.                 //  同伴还在，拒绝请求。
  110.                 exception = TRUE;
  111.         }
  112.     }
  113.     return exception;
  114. }
  117. int main (int argc, char *argv [])
  118. {
  119.     //  命令行参数可以为：
  120.     //      -p  作为主机启动, at tcp://localhost:5001
  121.     //      -b  作为备机启动, at tcp://localhost:5002
  122.     zctx_t *ctx = zctx_new ();
  123.     void *statepub = zsocket_new (ctx, ZMQ_PUB);
  124.     void *statesub = zsocket_new (ctx, ZMQ_SUB);
  125.     void *frontend = zsocket_new (ctx, ZMQ_ROUTER);
  126.     bstar_t fsm = { 0 };
  128.     if (argc == 2 && streq (argv [1], "-p")) {
  129.         printf ("I: 主机master，等待备机（slave）连接。\n");
  130.         zsocket_bind (frontend, "tcp://*:5001");
  131.         zsocket_bind (statepub, "tcp://*:5003");
  132.         zsocket_connect (statesub, "tcp://localhost:5004");
  133.         fsm.state = STATE_PRIMARY;
  134.     }
  135.     else
  136.     if (argc == 2 && streq (argv [1], "-b")) {
  137.         printf ("I: 备机slave，等待主机（master）连接。\n");
  138.         zsocket_bind (frontend, "tcp://*:5002");
  139.         zsocket_bind (statepub, "tcp://*:5004");
  140.         zsocket_connect (statesub, "tcp://localhost:5003");
  141.         fsm.state = STATE_BACKUP;
  142.     }
  143.     else {
  144.         printf ("Usage: bstarsrv { -p | -b }\n");
  145.         zctx_destroy (&ctx);
  146.         exit (0);
  147.     }
  148.     //  设定下一次发送状态的时间
  149.     int64_t send_state_at = zclock_time () + HEARTBEAT;
  151.     while (!zctx_interrupted) {
  152.         zmq_pollitem_t items [] = {
  153.             { frontend, 0, ZMQ_POLLIN, 0 },
  154.             { statesub, 0, ZMQ_POLLIN, 0 }
  155.         };
  156.         int time_left = (int) ((send_state_at - zclock_time ()));
  157.         if (time_left < 0)
  158.             time_left = 0;
  159.         int rc = zmq_poll (items, 2, time_left * ZMQ_POLL_MSEC);
  160.         if (rc == -1)
  161.             break;              //  上下文对象被关闭
  163.         if (items [0].revents & ZMQ_POLLIN) {
  164.             //  收到客户端请求
  165.             zmsg_t *msg = zmsg_recv (frontend);
  166.             fsm.event = CLIENT_REQUEST;
  167.             if (s_state_machine (&fsm) == FALSE)
  168.                 //  返回应答
  169.                 zmsg_send (&msg, frontend);
  170.             else
  171.                 zmsg_destroy (&msg);
  172.         }
  173.         if (items [1].revents & ZMQ_POLLIN) {
  174.             //  收到状态消息，作为事件处理
  175.             char *message = zstr_recv (statesub);
  176.             fsm.event = atoi (message);
  177.             free (message);
  178.             if (s_state_machine (&fsm))
  179.                 break;          //  错误，退出。
  180.             fsm.peer_expiry = zclock_time () + 2 * HEARTBEAT;
  181.         }
  182.         //  定时发送状态信息
  183.         if (zclock_time () >= send_state_at) {
  184.             char message [2];
  185.             sprintf (message, "%d", fsm.state);
  186.             zstr_send (statepub, message);
  187.             send_state_at = zclock_time () + HEARTBEAT;
  188.         }
  189.     }
  190.     if (zctx_interrupted)
  191.         printf ("W: 中断\n");
  193.     //  关闭套接字和上下文
  194.     zctx_destroy (&ctx);
  195.     return 0;
  196. }

下面是客户端代码：

bstarcli: Binary Star client in C

  1. //
  2. //  双子星模式 - 客户端
  3. //
  4. #include "czmq.h"
  6. #define REQUEST_TIMEOUT     1000    //  毫秒
  7. #define SETTLE_DELAY        2000    //  超时时间
  9. int main (void)
  10. {
  11.     zctx_t *ctx = zctx_new ();
  13.     char *server [] = { "tcp://localhost:5001", "tcp://localhost:5002" };
  14.     uint server_nbr = 0;
  16.     printf ("I: 正在连接服务器 %s...\n", server [server_nbr]);
  17.     void *client = zsocket_new (ctx, ZMQ_REQ);
  18.     zsocket_connect (client, server [server_nbr]);
  20.     int sequence = 0;
  21.     while (!zctx_interrupted) {
  22.         //  发送请求并等待应答
  23.         char request [10];
  24.         sprintf (request, "%d", ++sequence);
  25.         zstr_send (client, request);
  27.         int expect_reply = 1;
  28.         while (expect_reply) {
  29.             //  轮询套接字
  30.             zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } };
  31.             int rc = zmq_poll (items, 1, REQUEST_TIMEOUT * ZMQ_POLL_MSEC);
  32.             if (rc == -1)
  33.                 break;          //  中断
  35.             //  处理应答
  36.             if (items [0].revents & ZMQ_POLLIN) {
  37.                 //  审核应答编号
  38.                 char *reply = zstr_recv (client);
  39.                 if (atoi (reply) == sequence) {
  40.                     printf ("I: 服务端应答正常 (%s)\n", reply);
  41.                     expect_reply = 0;
  42.                     sleep (1);  //  每秒发送一个请求
  43.                 }
  44.                 else {
  45.                     printf ("E: 错误的应答内容: %s\n",
  46.                         reply);
  47.                 }
  48.                 free (reply);
  49.             }
  50.             else {
  51.                 printf ("W: 服务器无响应，正在重试\n");
  52.                 //  重开套接字
  53.                 zsocket_destroy (ctx, client);
  54.                 server_nbr = (server_nbr + 1) % 2;
  55.                 zclock_sleep (SETTLE_DELAY);
  56.                 printf ("I: 正在连接服务端 %s...\n",
  57.                         server [server_nbr]);
  58.                 client = zsocket_new (ctx, ZMQ_REQ);
  59.                 zsocket_connect (client, server [server_nbr]);
  61.                 //  使用新套接字重发请求
  62.                 zstr_send (client, request);
  63.             }
  64.         }
  65.     }
  66.     zctx_destroy (&ctx);
  67.     return 0;
  68. }

运行以下命令进行测试，顺序随意：

  1. bstarsrv -p     # Start primary
  2. bstarsrv -b     # Start backup
  3. bstarcli

可以将主机进程杀掉，测试故障恢复机制；再开启主机，杀掉备机，查看还原机制。要注意是由客户端触发这两个事件的。

下图展现了服务进程的状态图。绿色状态下会接收客户端请求，粉色状态会拒绝请求。事件指的是同伴的状态，所以“同伴激活态”指的是同伴机器告知我们它处于激活态。“客户请求”表示我们从客户端获得了请求，“客户投票”则指我们从客户端获得了请求并且同伴已经超时死亡。

8

需要注意的是，服务进程使用PUB-SUB套接字进行状态交换，其它类型的套接字在这里不适用。比如，PUSH和DEALER套接字在没有节点相连的时候会发生阻塞；PAIR套接字不会在节点断开后进行重连；ROUTER套接字需要地址才能发送消息。

These are the main limitations of the Binary Star pattern:

  • A server process cannot be part of more than one Binary Star pair.
  • A primary server can have a single backup server, no more.
  • The backup server cannot do useful work while in slave mode.
  • The backup server must be capable of handling full application loads.
  • Failover configuration cannot be modified at runtime.
  • Client applications must do some work to benefit from failover.

双子星反应堆

我们可以将双子星模式打包成一个类似反应堆的类，供以后复用。在C语言中，我们使用czmq的zloop类，其他语言应该会有相应的实现。以下是C语言版的bstar接口：

  1. // 创建双子星模式实例，使用本地（绑定）和远程（连接）端点来设置节点对。
  2. bstar_t *bstar_new (int primary, char *local, char *remote);
  4. // 销毁实例
  5. void bstar_destroy (bstar_t **self_p);
  7. // 返回底层的zloop反应堆，用以添加定时器、读取器、注册和取消等功能。
  8. zloop_t *bstar_zloop (bstar_t *self);
  10. // 注册投票读取器
  11. int bstar_voter (bstar_t *self, char *endpoint, int type,
  12. zloop_fn handler, void *arg);
  14. // 注册状态机处理器
  15. void bstar_new_master (bstar_t *self, zloop_fn handler, void *arg);
  16. void bstar_new_slave (bstar_t *self, zloop_fn handler, void *arg);
  18. // 开启反应堆，当回调函数返回-1，或进程收到SIGINT、SIGTERM信号时中止。
  19. int bstar_start (bstar_t *self);

以下是类的实现：

bstar: Binary Star core class in C

  1. /*  =====================================================================
  2.     bstar - Binary Star reactor
  4.     ---------------------------------------------------------------------
  5.     Copyright (c) 1991-2011 iMatix Corporation <www.imatix.com>
  6.     Copyright other contributors as noted in the AUTHORS file.
  8.     This file is part of the ZeroMQ Guide: http://zguide.zeromq.org
  10.     This is free software; you can redistribute it and/or modify it under
  11.     the terms of the GNU Lesser General Public License as published by
  12.     the Free Software Foundation; either version 3 of the License, or (at
  13.     your option) any later version.
  15.     This software is distributed in the hope that it will be useful, but
  16.     WITHOUT ANY WARRANTY; without even the implied warranty of
  17.     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  18.     Lesser General Public License for more details.
  20.     You should have received a copy of the GNU Lesser General Public
  21.     License along with this program. If not, see
  22.     <http://www.gnu.org/licenses/>.
  23.     =====================================================================
  24. */
  26. #include "bstar.h"
  28. //  服务器状态枚举
  29. typedef enum {
  30.     STATE_PRIMARY = 1,          //  主机，等待同伴连接
  31.     STATE_BACKUP = 2,           //  备机，等待同伴连接
  32.     STATE_ACTIVE = 3,           //  激活态，处理应用程序请求
  33.     STATE_PASSIVE = 4           //  被动态，不接收请求
  34. } state_t;
  36. //  对话节点事件
  37. typedef enum {
  38.     PEER_PRIMARY = 1,           //  主机
  39.     PEER_BACKUP = 2,            //  备机
  40.     PEER_ACTIVE = 3,            //  激活态
  41.     PEER_PASSIVE = 4,           //  被动态
  42.     CLIENT_REQUEST = 5          //  客户端请求
  43. } event_t;
  46. //  发送状态信息的间隔时间
  47. //  如果对方在两次心跳过后都没有应答，则视为断开
  48. #define BSTAR_HEARTBEAT     1000        //  In msecs
  50. //  类结构
  52. struct _bstar_t {
  53.     zctx_t *ctx;                //  私有上下文
  54.     zloop_t *loop;              //  反应堆循环
  55.     void *statepub;             //  状态发布者
  56.     void *statesub;             //  状态订阅者
  57.     state_t state;              //  当前状态
  58.     event_t event;              //  当前事件
  59.     int64_t peer_expiry;        //  判定节点死亡的时限
  60.     zloop_fn *voter_fn;         //  投票套接字处理器
  61.     void *voter_arg;            //  投票处理程序的参数
  62.     zloop_fn *master_fn;        //  成为master时回调
  63.     void *master_arg;           //  参数
  64.     zloop_fn *slave_fn;         //  成为slave时回调
  65.     void *slave_arg;            //  参数
  66. };
  69. //  ---------------------------------------------------------------------
  70. //  执行有限状态机（将事件绑定至状态）；
  71. //  发生异常时返回-1，正确时返回0。
  73. static int
  74. s_execute_fsm (bstar_t *self)
  75. {
  76.     int rc = 0;
  77.     //  主机等待同伴连接
  78.     //  该状态下接收CLIENT_REQUEST事件
  79.     if (self->state == STATE_PRIMARY) {
  80.         if (self->event == PEER_BACKUP) {
  81.             zclock_log ("I: 已连接至备机（slave），可以作为master运行。");
  82.             self->state = STATE_ACTIVE;
  83.             if (self->master_fn)
  84.                 (self->master_fn) (self->loop, NULL, self->master_arg);
  85.         }
  86.         else
  87.         if (self->event == PEER_ACTIVE) {
  88.             zclock_log ("I: 已连接至备机（master），可以作为slave运行。");
  89.             self->state = STATE_PASSIVE;
  90.             if (self->slave_fn)
  91.                 (self->slave_fn) (self->loop, NULL, self->slave_arg);
  92.         }
  93.         else
  94.         if (self->event == CLIENT_REQUEST) {
  95.             zclock_log ("I: 收到客户端请求，可作为master运行。");
  96.             self->state = STATE_ACTIVE;
  97.             if (self->master_fn)
  98.                 (self->master_fn) (self->loop, NULL, self->master_arg);
  99.         }
  100.     }
  101.     else
  102.     //  备机等待同伴连接
  103.     //  该状态下拒绝CLIENT_REQUEST事件
  104.     if (self->state == STATE_BACKUP) {
  105.         if (self->event == PEER_ACTIVE) {
  106.             zclock_log ("I: 已连接至主机（master），可以作为slave运行。");
  107.             self->state = STATE_PASSIVE;
  108.             if (self->slave_fn)
  109.                 (self->slave_fn) (self->loop, NULL, self->slave_arg);
  110.         }
  111.         else
  112.         if (self->event == CLIENT_REQUEST)
  113.             rc = -1;
  114.     }
  115.     else
  116.     //  服务器处于激活态
  117.     //  该状态下接受CLIENT_REQUEST事件
  118.     //  只有服务器死亡才会离开激活态
  119.     if (self->state == STATE_ACTIVE) {
  120.         if (self->event == PEER_ACTIVE) {
  121.             //  若出现两台master，则抛出异常
  122.             zclock_log ("E: 严重错误：双master。正在退出。");
  123.             rc = -1;
  124.         }
  125.     }
  126.     else
  127.     //  服务器处于被动态
  128.     //  若同伴已死，CLIENT_REQUEST事件将触发故障恢复
  129.     if (self->state == STATE_PASSIVE) {
  130.         if (self->event == PEER_PRIMARY) {
  131.             //  同伴正在重启 - 转为激活态，同伴将转为被动态。
  132.             zclock_log ("I: 主机（slave）正在重启，可作为master运行。");
  133.             self->state = STATE_ACTIVE;
  134.         }
  135.         else
  136.         if (self->event == PEER_BACKUP) {
  137.             //  同伴正在重启 - 转为激活态，同伴将转为被动态。
  138.             zclock_log ("I: 备机（slave）正在重启，可作为master运行。");
  139.             self->state = STATE_ACTIVE;
  140.         }
  141.         else
  142.         if (self->event == PEER_PASSIVE) {
  143.             //  若出现两台slave，集群将无响应
  144.             zclock_log ("E: 严重错误：双slave。正在退出");
  145.             rc = -1;
  146.         }
  147.         else
  148.         if (self->event == CLIENT_REQUEST) {
  149.             //  若心跳超时，同伴将成为master；
  150.             //  此行为由客户端请求触发。
  151.             assert (self->peer_expiry > 0);
  152.             if (zclock_time () >= self->peer_expiry) {
  153.                 //  同伴已死，转为激活态。
  154.                 zclock_log ("I: 故障恢复，可作为master运行。");
  155.                 self->state = STATE_ACTIVE;
  156.             }
  157.             else
  158.                 //  同伴还在，拒绝请求。
  159.                 rc = -1;
  160.         }
  161.         //  触发状态更改事件处理函数
  162.         if (self->state == STATE_ACTIVE && self->master_fn)
  163.             (self->master_fn) (self->loop, NULL, self->master_arg);
  164.     }
  165.     return rc;
  166. }
  169. //  ---------------------------------------------------------------------
  170. //  反应堆事件处理程序
  172. //  发送状态信息
  173. int s_send_state (zloop_t *loop, void *socket, void *arg)
  174. {
  175.     bstar_t *self = (bstar_t *) arg;
  176.     zstr_sendf (self->statepub, "%d", self->state);
  177.     return 0;
  178. }
  180. //  接收状态信息，启动有限状态机
  181. int s_recv_state (zloop_t *loop, void *socket, void *arg)
  182. {
  183.     bstar_t *self = (bstar_t *) arg;
  184.     char *state = zstr_recv (socket);
  185.     if (state) {
  186.         self->event = atoi (state);
  187.         self->peer_expiry = zclock_time () + 2 * BSTAR_HEARTBEAT;
  188.         free (state);
  189.     }
  190.     return s_execute_fsm (self);
  191. }
  193. //  收到应用程序请求，判断是否接收
  194. int s_voter_ready (zloop_t *loop, void *socket, void *arg)
  195. {
  196.     bstar_t *self = (bstar_t *) arg;
  197.     //  如果能够处理请求，则调用函数
  198.     self->event = CLIENT_REQUEST;
  199.     if (s_execute_fsm (self) == 0) {
  200.         puts ("CLIENT REQUEST");
  201.         (self->voter_fn) (self->loop, socket, self->voter_arg);
  202.     }
  203.     else {
  204.         //  销毁等待中的消息
  205.         zmsg_t *msg = zmsg_recv (socket);
  206.         zmsg_destroy (&msg);
  207.     }
  208.     return 0;
  209. }
  212. //  ---------------------------------------------------------------------
  213. //  构造函数
  215. bstar_t *
  216. bstar_new (int primary, char *local, char *remote)
  217. {
  218.     bstar_t
  219.         *self;
  221.     self = (bstar_t *) zmalloc (sizeof (bstar_t));
  223.     //  初始化双子星
  224.     self->ctx = zctx_new ();
  225.     self->loop = zloop_new ();
  226.     self->state = primary? STATE_PRIMARY: STATE_BACKUP;
  228.     //  创建状态PUB套接字
  229.     self->statepub = zsocket_new (self->ctx, ZMQ_PUB);
  230.     zsocket_bind (self->statepub, local);
  232.     //  创建状态SUB套接字
  233.     self->statesub = zsocket_new (self->ctx, ZMQ_SUB);
  234.     zsocket_connect (self->statesub, remote);
  236.     //  设置基本的反应堆事件处理器
  237.     zloop_timer (self->loop, BSTAR_HEARTBEAT, 0, s_send_state, self);
  238.     zloop_reader (self->loop, self->statesub, s_recv_state, self);
  239.     return self;
  240. }
  243. //  ---------------------------------------------------------------------
  244. //  析构函数
  246. void
  247. bstar_destroy (bstar_t **self_p)
  248. {
  249.     assert (self_p);
  250.     if (*self_p) {
  251.         bstar_t *self = *self_p;
  252.         zloop_destroy (&self->loop);
  253.         zctx_destroy (&self->ctx);
  254.         free (self);
  255.         *self_p = NULL;
  256.     }
  257. }
  260. //  ---------------------------------------------------------------------
  261. //  返回底层zloop对象，用以添加额外的定时器、阅读器等。
  263. zloop_t *
  264. bstar_zloop (bstar_t *self)
  265. {
  266.     return self->loop;
  267. }
  270. //  ---------------------------------------------------------------------
  271. //  创建套接字，连接至本地端点，注册成为阅读器；
  272. //  只有当有限状态机允许时才会读取该套接字；
  273. //  从该套接字获得的消息将作为一次“投票”；
  274. //  我们要求双子星模式中只有一个“投票”套接字。
  276. int
  277. bstar_voter (bstar_t *self, char *endpoint, int type, zloop_fn handler,
  278.              void *arg)
  279. {
  280.     //  保存原始的回调函数和参数，稍后使用
  281.     void *socket = zsocket_new (self->ctx, type);
  282.     zsocket_bind (socket, endpoint);
  283.     assert (!self->voter_fn);
  284.     self->voter_fn = handler;
  285.     self->voter_arg = arg;
  286.     return zloop_reader (self->loop, socket, s_voter_ready, self);
  287. }
  289. //  ---------------------------------------------------------------------
  290. //  注册状态变化事件处理器
  292. void
  293. bstar_new_master (bstar_t *self, zloop_fn handler, void *arg)
  294. {
  295.     assert (!self->master_fn);
  296.     self->master_fn = handler;
  297.     self->master_arg = arg;
  298. }
  300. void
  301. bstar_new_slave (bstar_t *self, zloop_fn handler, void *arg)
  302. {
  303.     assert (!self->slave_fn);
  304.     self->slave_fn = handler;
  305.     self->slave_arg = arg;
  306. }
  309. //  ---------------------------------------------------------------------
  310. //  启用或禁止跟踪信息
  311. void bstar_set_verbose (bstar_t *self, Bool verbose)
  312. {
  313.     zloop_set_verbose (self->loop, verbose);
  314. }
  317. //  ---------------------------------------------------------------------
  318. //  开启反应堆，当回调函数返回-1，或进程收到SIGINT、SIGTERM信号时中止。
  320. int
  321. bstar_start (bstar_t *self)
  322. {
  323.     assert (self->voter_fn);
  324.     return zloop_start (self->loop);
  325. }

这样一来，我们的服务端代码会变得非常简短：

bstarsrv2: Binary Star server, using core class in C

  1. //
  2. //  双子星模式服务端，使用bstar反应堆
  3. //
  5. //  直接编译，不建类库
  6. #include "bstar.c"
  8. //  Echo service
  9. int s_echo (zloop_t *loop, void *socket, void *arg)
  10. {
  11.     zmsg_t *msg = zmsg_recv (socket);
  12.     zmsg_send (&msg, socket);
  13.     return 0;
  14. }
  16. int main (int argc, char *argv [])
  17. {
  18.     //  命令行参数可以为：
  19.     //      -p  作为主机启动, at tcp://localhost:5001
  20.     //      -b  作为备机启动, at tcp://localhost:5002
  21.     bstar_t *bstar;
  22.     if (argc == 2 && streq (argv [1], "-p")) {
  23.         printf ("I: 主机master，等待备机（slave）连接。\n");
  24.         bstar = bstar_new (BSTAR_PRIMARY,
  25.             "tcp://*:5003", "tcp://localhost:5004");
  26.         bstar_voter (bstar, "tcp://*:5001", ZMQ_ROUTER, s_echo, NULL);
  27.     }
  28.     else
  29.     if (argc == 2 && streq (argv [1], "-b")) {
  30.         printf ("I: 备机slave，等待主机（master）连接。\n");
  31.         bstar = bstar_new (BSTAR_BACKUP,
  32.             "tcp://*:5004", "tcp://localhost:5003");
  33.         bstar_voter (bstar, "tcp://*:5002", ZMQ_ROUTER, s_echo, NULL);
  34.     }
  35.     else {
  36.         printf ("Usage: bstarsrvs { -p | -b }\n");
  37.         exit (0);
  38.     }
  39.     bstar_start (bstar);
  40.     bstar_destroy (&bstar);
  41.     return 0;
  42. }

无中间件的可靠性（自由者模式）

我们讲了那么多关于中间件的示例，好像有些违背“ZMQ是无中间件”的说法。但要知道在现实生活中，中间件一直是让人又爱又恨的东西。实践中的很多消息架构能都在使用中间件进行分布式架构的搭建，所以说最终的决定还是需要你自己去权衡的。这也是为什么虽然我能驾车10分钟到一个大型商场里购买五箱音量，但我还是会选择走10分钟到楼下的便利店里去买。这种出于经济方面的考虑（时间、精力、成本等）不仅在日常生活中很常见，在软件架构中也很重要。

这就是为什么ZMQ不会强制使用带有中间件的架构，但仍提供了像内置装置这样的中间件供编程人员自由选用。

这一节我们会打破以往使用中间件进行可靠性设计的架构，转而使用点对点架构，即自由者模式，来进行可靠的消息传输。我们的示例程序会是一个名称解析服务。ZMQ中的一个常见问题是：我们如何得知需要连接的端点？在代码中直接写入TCP/IP地址肯定是不合适的；使用配置文件会造成管理上的不便。试想一下，你要在上百台计算机中进行配置，只是为了让它们知道google.com的IP地址是74.125.230.82。

一个ZMQ的名称解析服务需要实现的功能有：

  • 将逻辑名称解析为一个或多个端点地址，包括绑定端和连接端。实际使用时，名称服务会提供一组端点。
  • 允许我们在不同的环境下，即开发环境和生产环境，进行解析；
  • 该服务必须是可靠的，否则应用程序将无法连接到网络。

为管家模式提供名称解析服务会很有用，虽然将代理程序的端点对外暴露也很简单，但是如果用好名称解析服务，那它将成为唯一一个对外暴露的接口，将更便于管理。

我们需要处理的故障类型有：服务崩溃或重启、服务过载、网络因素等。为获取可靠性，我们必须建立一个服务群，当某个服务端崩溃后，客户端可以连接其他的服务端。实践中，两个服务端就已经足够了，但事实上服务端的数量可以是任意个。

9

在这个架构中，大量客户端和少量服务端进行通信，服务端将套接字绑定至单独的端口，这和管家模式中的代理有很大不同。对于客户端来说，它有这样几种选择：

  • 客户端可以使用REQ套接字和懒惰海盗模式，但需要有一个机制防止客户端不断地请求已停止的服务端。

  • 客户端可以使用DEALER套接字，向所有的服务端发送请求。很简单，但并不太妙；

  • 客户端使用ROUTER套接字，连接特定的服务端。但客户端如何得知服务端的套接字标识呢？一种方式是让服务端主动连接客户端（很复杂），或者将服务端标识写入代码进行固化（很混乱）。

模型一：简单重试

让我们先尝试简单的方案，重写懒惰海盗模式，让其能够和多个服务端进行通信。启动服务端时用命令行参数指定端口。然后启动多个服务端。

flserver1: Freelance server, Model One in C

  1. //
  2. //  自由者模式 - 服务端 - 模型1
  3. //  提供echo服务
  4. //
  5. #include "czmq.h"
  7. int main (int argc, char *argv [])
  8. {
  9.     if (argc < 2) {
  10.         printf ("I: syntax: %s <endpoint>\n", argv [0]);
  11.         exit (EXIT_SUCCESS);
  12.     }
  13.     zctx_t *ctx = zctx_new ();
  14.     void *server = zsocket_new (ctx, ZMQ_REP);
  15.     zsocket_bind (server, argv [1]);
  17.     printf ("I: echo服务端点： %s\n", argv [1]);
  18.     while (TRUE) {
  19.         zmsg_t *msg = zmsg_recv (server);
  20.         if (!msg)
  21.             break;          //  中断
  22.         zmsg_send (&msg, server);
  23.     }
  24.     if (zctx_interrupted)
  25.         printf ("W: 中断\n");
  27.     zctx_destroy (&ctx);
  28.     return 0;
  29. }

启动客户端，指定一个或多个端点：

flclient1: Freelance client, Model One in C

  1. //
  2. //  自由者模式 - 客户端 - 模型1
  3. //  使用REQ套接字请求一个或多个服务端
  4. //
  5. #include "czmq.h"
  7. #define REQUEST_TIMEOUT     1000
  8. #define MAX_RETRIES         3       //  尝试次数
  11. static zmsg_t *
  12. s_try_request (zctx_t *ctx, char *endpoint, zmsg_t *request)
  13. {
  14.     printf ("I: 在端点 %s 上尝试请求echo服务...\n", endpoint);
  15.     void *client = zsocket_new (ctx, ZMQ_REQ);
  16.     zsocket_connect (client, endpoint);
  18.     //  发送请求，并等待应答
  19.     zmsg_t *msg = zmsg_dup (request);
  20.     zmsg_send (&msg, client);
  21.     zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } };
  22.     zmq_poll (items, 1, REQUEST_TIMEOUT * ZMQ_POLL_MSEC);
  23.     zmsg_t *reply = NULL;
  24.     if (items [0].revents & ZMQ_POLLIN)
  25.         reply = zmsg_recv (client);
  27.     //  关闭套接字
  28.     zsocket_destroy (ctx, client);
  29.     return reply;
  30. }
  33. int main (int argc, char *argv [])
  34. {
  35.     zctx_t *ctx = zctx_new ();
  36.     zmsg_t *request = zmsg_new ();
  37.     zmsg_addstr (request, "Hello world");
  38.     zmsg_t *reply = NULL;
  40.     int endpoints = argc - 1;
  41.     if (endpoints == 0)
  42.         printf ("I: syntax: %s <endpoint> ...\n", argv [0]);
  43.     else
  44.     if (endpoints == 1) {
  45.         //  若只有一个端点，则尝试N次
  46.         int retries;
  47.         for (retries = 0; retries < MAX_RETRIES; retries++) {
  48.             char *endpoint = argv [1];
  49.             reply = s_try_request (ctx, endpoint, request);
  50.             if (reply)
  51.                 break;          //  成功
  52.             printf ("W: 没有收到 %s 的应答, 准备重试...\n", endpoint);
  53.         }
  54.     }
  55.     else {
  56.         //  若有多个端点，则每个尝试一次
  57.         int endpoint_nbr;
  58.         for (endpoint_nbr = 0; endpoint_nbr < endpoints; endpoint_nbr++) {
  59.             char *endpoint = argv [endpoint_nbr + 1];
  60.             reply = s_try_request (ctx, endpoint, request);
  61.             if (reply)
  62.                 break;          //  Successful
  63.             printf ("W: 没有收到 %s 的应答\n", endpoint);
  64.         }
  65.     }
  66.     if (reply)
  67.         printf ("服务运作正常\n");
  69.     zmsg_destroy (&request);
  70.     zmsg_destroy (&reply);
  71.     zctx_destroy (&ctx);
  72.     return 0;
  73. }

可用如下命令运行：

  1. flserver1 tcp://*:5555 &
  2. flserver1 tcp://*:5556 &
  3. flclient1 tcp://localhost:5555 tcp://localhost:5556

客户端的核心机制是懒惰海盗模式，即获得一次成功的应答后就结束。会有两种情况：

  • 如果只有一个服务端，客户端会再尝试N次后停止，这和懒惰海盗模式的逻辑一致；
  • 如果有多个服务端，客户端会每个尝试一次，收到应答后停止。

这种机制补充了海盗模式，使其能够克服只有一个服务端的情况。

但是，这种设计无法在现实程序中使用：当有很多客户端连接了服务端，而主服务端崩溃了，那所有客户端都需要在超时后才能继续执行。

模型二：批量发送

下面让我们使用DEALER套接字。我们的目标是能再最短的时间里收到一个应答，不能受主服务端崩溃的影响。可以采取以下措施：

  • 连接所有的服务端；
  • 当有请求时，一次性发送给所有的服务端；
  • 等待第一个应答；
  • 忽略其他应答。

这样设计客户端时，当发送请求后，所有的服务端都会收到这个请求，并返回应答。如果某个服务端断开连接了，ZMQ可能会将请求发给其他服务端，导致某些服务端会收到两次请求。

更麻烦的是客户端无法得知应答的数量，容易发生混乱。

我们可以为请求进行编号，忽略不匹配的应答。我们要对服务端进行改造，返回的消息中需要包含请求编号：
flserver2: Freelance server, Model Two in C

  1. //
  2. //  自由者模式 - 服务端 - 模型2
  3. //  返回带有请求编号的OK信息
  4. //
  5. #include "czmq.h"
  7. int main (int argc, char *argv [])
  8. {
  9.     if (argc < 2) {
  10.         printf ("I: syntax: %s <endpoint>\n", argv [0]);
  11.         exit (EXIT_SUCCESS);
  12.     }
  13.     zctx_t *ctx = zctx_new ();
  14.     void *server = zsocket_new (ctx, ZMQ_REP);
  15.     zsocket_bind (server, argv [1]);
  17.     printf ("I: 服务已就绪 %s\n", argv [1]);
  18.     while (TRUE) {
  19.         zmsg_t *request = zmsg_recv (server);
  20.         if (!request)
  21.             break;          //  中断
  22.         //  判断请求内容是否正确
  23.         assert (zmsg_size (request) == 2);
  25.         zframe_t *address = zmsg_pop (request);
  26.         zmsg_destroy (&request);
  28.         zmsg_t *reply = zmsg_new ();
  29.         zmsg_add (reply, address);
  30.         zmsg_addstr (reply, "OK");
  31.         zmsg_send (&reply, server);
  32.     }
  33.     if (zctx_interrupted)
  34.         printf ("W: interrupted\n");
  36.     zctx_destroy (&ctx);
  37.     return 0;
  38. }

客户端代码：

flclient2: Freelance client, Model Two in C

  1. //
  2. //  自由者模式 - 客户端 - 模型2
  3. //  使用DEALER套接字发送批量消息
  4. //
  5. #include "czmq.h"
  7. //  超时时间
  8. #define GLOBAL_TIMEOUT 2500
  10. //  将客户端API封装成一个类
  12. #ifdef __cplusplus
  13. extern "C" {
  14. #endif
  16. //  声明类结构
  17. typedef struct _flclient_t flclient_t;
  19. flclient_t *
  20.     flclient_new (void);
  21. void
  22.     flclient_destroy (flclient_t **self_p);
  23. void
  24.     flclient_connect (flclient_t *self, char *endpoint);
  25. zmsg_t *
  26.     flclient_request (flclient_t *self, zmsg_t **request_p);
  28. #ifdef __cplusplus
  29. }
  30. #endif
  33. int main (int argc, char *argv [])
  34. {
  35.     if (argc == 1) {
  36.         printf ("I: syntax: %s <endpoint> ...\n", argv [0]);
  37.         exit (EXIT_SUCCESS);
  38.     }
  39.     //  创建自由者模式客户端
  40.     flclient_t *client = flclient_new ();
  42.     //  连接至各个端点
  43.     int argn;
  44.     for (argn = 1; argn < argc; argn++)
  45.         flclient_connect (client, argv [argn]);
  47.     //  发送一组请求，并记录时间
  48.     int requests = 10000;
  49.     uint64_t start = zclock_time ();
  50.     while (requests--) {
  51.         zmsg_t *request = zmsg_new ();
  52.         zmsg_addstr (request, "random name");
  53.         zmsg_t *reply = flclient_request (client, &request);
  54.         if (!reply) {
  55.             printf ("E: 名称解析服务不可用，正在退出\n");
  56.             break;
  57.         }
  58.         zmsg_destroy (&reply);
  59.     }
  60.     printf ("平均请求时间: %d 微秒\n",
  61.         (int) (zclock_time () - start) / 10);
  63.     flclient_destroy (&client);
  64.     return 0;
  65. }
  69. //  --------------------------------------------------------------------
  70. //  类结构
  72. struct _flclient_t {
  73.     zctx_t *ctx;        //  上下文
  74.     void *socket;       //  用于和服务端通信的DEALER套接字
  75.     size_t servers;     //  以连接的服务端数量
  76.     uint sequence;      //  已发送的请求数
  77. };
  80. //  --------------------------------------------------------------------
  81. //  Constructor
  83. flclient_t *
  84. flclient_new (void)
  85. {
  86.     flclient_t
  87.         *self;
  89.     self = (flclient_t *) zmalloc (sizeof (flclient_t));
  90.     self->ctx = zctx_new ();
  91.     self->socket = zsocket_new (self->ctx, ZMQ_DEALER);
  92.     return self;
  93. }
  95. //  --------------------------------------------------------------------
  96. //  析构函数
  98. void
  99. flclient_destroy (flclient_t **self_p)
  100. {
  101.     assert (self_p);
  102.     if (*self_p) {
  103.         flclient_t *self = *self_p;
  104.         zctx_destroy (&self->ctx);
  105.         free (self);
  106.         *self_p = NULL;
  107.     }
  108. }
  110. //  --------------------------------------------------------------------
  111. //  连接至新的服务端端点
  113. void
  114. flclient_connect (flclient_t *self, char *endpoint)
  115. {
  116.     assert (self);
  117.     zsocket_connect (self->socket, endpoint);
  118.     self->servers++;
  119. }
  121. //  --------------------------------------------------------------------
  122. //  发送请求，接收应答
  123. //  发送后销毁请求
  125. zmsg_t *
  126. flclient_request (flclient_t *self, zmsg_t **request_p)
  127. {
  128.     assert (self);
  129.     assert (*request_p);
  130.     zmsg_t *request = *request_p;
  132.     //  向消息添加编号和空帧
  133.     char sequence_text [10];
  134.     sprintf (sequence_text, "%u", ++self->sequence);
  135.     zmsg_pushstr (request, sequence_text);
  136.     zmsg_pushstr (request, "");
  138.     //  向所有已连接的服务端发送请求
  139.     int server;
  140.     for (server = 0; server < self->servers; server++) {
  141.         zmsg_t *msg = zmsg_dup (request);
  142.         zmsg_send (&msg, self->socket);
  143.     }
  144.     //  接收来自任何服务端的应答
  145.     //  因为我们可能poll多次，所以每次都进行计算
  146.     zmsg_t *reply = NULL;
  147.     uint64_t endtime = zclock_time () + GLOBAL_TIMEOUT;
  148.     while (zclock_time () < endtime) {
  149.         zmq_pollitem_t items [] = { { self->socket, 0, ZMQ_POLLIN, 0 } };
  150.         zmq_poll (items, 1, (endtime - zclock_time ()) * ZMQ_POLL_MSEC);
  151.         if (items [0].revents & ZMQ_POLLIN) {
  152.             //  应答内容是 [empty][sequence][OK]
  153.             reply = zmsg_recv (self->socket);
  154.             assert (zmsg_size (reply) == 3);
  155.             free (zmsg_popstr (reply));
  156.             char *sequence = zmsg_popstr (reply);
  157.             int sequence_nbr = atoi (sequence);
  158.             free (sequence);
  159.             if (sequence_nbr == self->sequence)
  160.                 break;
  161.         }
  162.     }
  163.     zmsg_destroy (request_p);
  164.     return reply;
  165. }

几点说明：

  • 客户端被封装成了一个API类，将复杂的代码都包装了起来。
  • 客户端会在几秒之后放弃寻找可用的服务端；
  • 客户端需要创建一个合法的REP信封，所以需要添加一个空帧。

程序中，客户端发出了1万次名称解析请求（虽然是假的），并计算平均耗费时间。在我的测试机上，有一个服务端时，耗时60微妙；三个时80微妙。

该模型的优缺点是：

  • 优点：简单，容易理解和编写；
  • 优点：它工作迅速，有重试机制；
  • 缺点：占用了额外的网络带宽；
  • 缺点：我们不能为服务端设置优先级，如主服务、次服务等；
  • 缺点：服务端不能同时处理多个请求。

Model Three - Complex and Nasty

批量发送模型看起来不太真实，那就让我们来探索最后这个极度复杂的模型。很有可能在编写完之后我们又会转而使用批量发送，哈哈，这就是我的作风。

我们可以将客户端使用的套接字更换为ROUTER，让我们能够向特定的服务端发送请求，停止向已死亡的服务端发送请求，从而做得尽可能地智能。我们还可以将服务端的套接字更换为ROUTER，从而突破单线程的瓶颈。

但是，使用ROUTER-ROUTER套接字连接两个瞬时套接字是不可行的，节点只有在收到第一条消息时才会为对方生成套接字标识。唯一的方法是让其中一个节点使用持久化的套接字，比较好的方式是让客户端知道服务端的标识，即服务端作为持久化的套接字。

为了避免产生新的配置项，我们直接使用服务端的端点作为套接字标识。

回想一下ZMQ套接字标识是如何工作的。服务端的ROUTER套接字为自己设置一个标识（在绑定之前），当客户端连接时，通过一个握手的过程来交换双方的标识。客户端的ROUTER套接字会先发送一条空消息，服务端为客户端生成一个随机的UUID。然后，服务端会向客户端发送自己的标识。

这样一来，客户端就可以将消息发送给特定的服务端了。不过还有一个问题：我们不知道服务端会在什么时候完成这个握手的过程。如果服务端是在线的，那可能几毫秒就能完成。如果不在线，那可能需要很久很久。

这里有一个矛盾：我们需要知道服务端何时连接成功且能够开始工作。自由者模式不像中间件模式，它的服务端必须要先发送请求后才能的应答。所以在服务端发送消息给客户端之前，客户端必须要先请求服务端，这看似是不可能的。

我有一个解决方法，那就是批量发送。这里发送的不是真正的请求，而是一个试探性的心跳（PING-PONG）。当收到应答时，就说明对方是在线的。

下面让我们制定一个协议，来定义自由者模式是如何传递这种心跳的：

实现这个协议的服务端很方便，下面就是经过改造的echo服务：

flserver3: Freelance server, Model Three in C

  1. //
  2. //  自由者模式 - 服务端 - 模型3
  3. //  使用ROUTER-ROUTER套接字进行通信；单线程。
  4. //
  5. #include "czmq.h"
  7. int main (int argc, char *argv [])
  8. {
  9.     int verbose = (argc > 1 && streq (argv [1], "-v"));
  11.     zctx_t *ctx = zctx_new ();
  13.     //  准备服务端套接字，其标识和端点名相同
  14.     char *bind_endpoint = "tcp://*:5555";
  15.     char *connect_endpoint = "tcp://localhost:5555";
  16.     void *server = zsocket_new (ctx, ZMQ_ROUTER);
  17.     zmq_setsockopt (server,
  18.         ZMQ_IDENTITY, connect_endpoint, strlen (connect_endpoint));
  19.     zsocket_bind (server, bind_endpoint);
  20.     printf ("I: 服务端已准备就绪 %s\n", bind_endpoint);
  22.     while (!zctx_interrupted) {
  23.         zmsg_t *request = zmsg_recv (server);
  24.         if (verbose && request)
  25.             zmsg_dump (request);
  26.         if (!request)
  27.             break;          //  中断
  29.         //  Frame 0: 客户端标识
  30.         //  Frame 1: 心跳，或客户端控制信息帧
  31.         //  Frame 2: 请求内容
  32.         zframe_t *address = zmsg_pop (request);
  33.         zframe_t *control = zmsg_pop (request);
  34.         zmsg_t *reply = zmsg_new ();
  35.         if (zframe_streq (control, "PONG"))
  36.             zmsg_addstr (reply, "PONG");
  37.         else {
  38.             zmsg_add (reply, control);
  39.             zmsg_addstr (reply, "OK");
  40.         }
  41.         zmsg_destroy (&request);
  42.         zmsg_push (reply, address);
  43.         if (verbose && reply)
  44.             zmsg_dump (reply);
  45.         zmsg_send (&reply, server);
  46.     }
  47.     if (zctx_interrupted)
  48.         printf ("W: 中断\n");
  50.     zctx_destroy (&ctx);
  51.     return 0;
  52. }

但是，自由者模式的客户端会变得大一写。为了清晰期间，我们将其拆分为两个类来实现。首先是在上层使用的程序：

flclient3: Freelance client, Model Three in C

  1. //
  2. //  自由者模式 - 客户端 - 模型3
  3. //  使用flcliapi类来封装自由者模式
  4. //
  5. //  直接编译，不建类库
  6. #include "flcliapi.c"
  8. int main (void)
  9. {
  10.     //  创建自由者模式实例
  11.     flcliapi_t *client = flcliapi_new ();
  13.     //  链接至服务器端点
  14.     flcliapi_connect (client, "tcp://localhost:5555");
  15.     flcliapi_connect (client, "tcp://localhost:5556");
  16.     flcliapi_connect (client, "tcp://localhost:5557");
  18.     //  发送随机请求，计算时间
  19.     int requests = 1000;
  20.     uint64_t start = zclock_time ();
  21.     while (requests--) {
  22.         zmsg_t *request = zmsg_new ();
  23.         zmsg_addstr (request, "random name");
  24.         zmsg_t *reply = flcliapi_request (client, &request);
  25.         if (!reply) {
  26.             printf ("E: 名称解析服务不可用，正在退出\n");
  27.             break;
  28.         }
  29.         zmsg_destroy (&reply);
  30.     }
  31.     printf ("平均执行时间： %d usec\n",
  32.         (int) (zclock_time () - start) / 10);
  34.     flcliapi_destroy (&client);
  35.     return 0;
  36. }

下面是该模式复杂的实现过程：

flcliapi: Freelance client API in C

  1. /*  =====================================================================
  2.     flcliapi - Freelance Pattern agent class
  3.     Model 3: uses ROUTER socket to address specific services
  5.     ---------------------------------------------------------------------
  6.     Copyright (c) 1991-2011 iMatix Corporation <www.imatix.com>
  7.     Copyright other contributors as noted in the AUTHORS file.
  9.     This file is part of the ZeroMQ Guide: http://zguide.zeromq.org
  11.     This is free software; you can redistribute it and/or modify it under
  12.     the terms of the GNU Lesser General Public License as published by
  13.     the Free Software Foundation; either version 3 of the License, or (at
  14.     your option) any later version.
  16.     This software is distributed in the hope that it will be useful, but
  17.     WITHOUT ANY WARRANTY; without even the implied warranty of
  18.     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  19.     Lesser General Public License for more details.
  21.     You should have received a copy of the GNU Lesser General Public
  22.     License along with this program. If not, see
  23.     <http://www.gnu.org/licenses/>.
  24.     =====================================================================
  25. */
  27. #include "flcliapi.h"
  29. //  请求超时时间
  30. #define GLOBAL_TIMEOUT  3000    //  msecs
  31. //  心跳间隔
  32. #define PING_INTERVAL   2000    //  msecs
  33. //  判定服务死亡的时间
  34. #define SERVER_TTL      6000    //  msecs
  37. //  =====================================================================
  38. //  同步部分，在应用程序层面运行
  40. //  ---------------------------------------------------------------------
  41. //  类结构
  43. struct _flcliapi_t {
  44.     zctx_t *ctx;        //  上下文
  45.     void *pipe;         //  用于和主线程通信的套接字
  46. };
  48. //  这是运行后台代理程序的线程
  49. static void flcliapi_agent (void *args, zctx_t *ctx, void *pipe);
  52. //  ---------------------------------------------------------------------
  53. //  构造函数
  55. flcliapi_t *
  56. flcliapi_new (void)
  57. {
  58.     flcliapi_t
  59.         *self;
  61.     self = (flcliapi_t *) zmalloc (sizeof (flcliapi_t));
  62.     self->ctx = zctx_new ();
  63.     self->pipe = zthread_fork (self->ctx, flcliapi_agent, NULL);
  64.     return self;
  65. }
  67. //  ---------------------------------------------------------------------
  68. //  析构函数
  70. void
  71. flcliapi_destroy (flcliapi_t **self_p)
  72. {
  73.     assert (self_p);
  74.     if (*self_p) {
  75.         flcliapi_t *self = *self_p;
  76.         zctx_destroy (&self->ctx);
  77.         free (self);
  78.         *self_p = NULL;
  79.     }
  80. }
  82. //  ---------------------------------------------------------------------
  83. //  连接至新服务器端点
  84. //  消息内容：[CONNECT][endpoint]
  86. void
  87. flcliapi_connect (flcliapi_t *self, char *endpoint)
  88. {
  89.     assert (self);
  90.     assert (endpoint);
  91.     zmsg_t *msg = zmsg_new ();
  92.     zmsg_addstr (msg, "CONNECT");
  93.     zmsg_addstr (msg, endpoint);
  94.     zmsg_send (&msg, self->pipe);
  95.     zclock_sleep (100);      //  等待连接
  96. }
  98. //  ---------------------------------------------------------------------
  99. //  发送并销毁请求，接收应答
  101. zmsg_t *
  102. flcliapi_request (flcliapi_t *self, zmsg_t **request_p)
  103. {
  104.     assert (self);
  105.     assert (*request_p);
  107.     zmsg_pushstr (*request_p, "REQUEST");
  108.     zmsg_send (request_p, self->pipe);
  109.     zmsg_t *reply = zmsg_recv (self->pipe);
  110.     if (reply) {
  111.         char *status = zmsg_popstr (reply);
  112.         if (streq (status, "FAILED"))
  113.             zmsg_destroy (&reply);
  114.         free (status);
  115.     }
  116.     return reply;
  117. }
  120. //  =====================================================================
  121. //  异步部分，在后台运行
  123. //  ---------------------------------------------------------------------
  124. //  单个服务端信息
  126. typedef struct {
  127.     char *endpoint;             //  服务端端点/套接字标识
  128.     uint alive;                 //  是否在线
  129.     int64_t ping_at;            //  下一次心跳时间
  130.     int64_t expires;            //  过期时间
  131. } server_t;
  133. server_t *
  134. server_new (char *endpoint)
  135. {
  136.     server_t *self = (server_t *) zmalloc (sizeof (server_t));
  137.     self->endpoint = strdup (endpoint);
  138.     self->alive = 0;
  139.     self->ping_at = zclock_time () + PING_INTERVAL;
  140.     self->expires = zclock_time () + SERVER_TTL;
  141.     return self;
  142. }
  144. void
  145. server_destroy (server_t **self_p)
  146. {
  147.     assert (self_p);
  148.     if (*self_p) {
  149.         server_t *self = *self_p;
  150.         free (self->endpoint);
  151.         free (self);
  152.         *self_p = NULL;
  153.     }
  154. }
  156. int
  157. server_ping (char *key, void *server, void *socket)
  158. {
  159.     server_t *self = (server_t *) server;
  160.     if (zclock_time () >= self->ping_at) {
  161.         zmsg_t *ping = zmsg_new ();
  162.         zmsg_addstr (ping, self->endpoint);
  163.         zmsg_addstr (ping, "PING");
  164.         zmsg_send (&ping, socket);
  165.         self->ping_at = zclock_time () + PING_INTERVAL;
  166.     }
  167.     return 0;
  168. }
  170. int
  171. server_tickless (char *key, void *server, void *arg)
  172. {
  173.     server_t *self = (server_t *) server;
  174.     uint64_t *tickless = (uint64_t *) arg;
  175.     if (*tickless > self->ping_at)
  176.         *tickless = self->ping_at;
  177.     return 0;
  178. }
  181. //  ---------------------------------------------------------------------
  182. //  后台处理程序信息
  184. typedef struct {
  185.     zctx_t *ctx;                //  上下文
  186.     void *pipe;                 //  用于应用程序通信的套接字
  187.     void *router;               //  用于服务端通信的套接字
  188.     zhash_t *servers;           //  已连接的服务端
  189.     zlist_t *actives;           //  在线的服务端
  190.     uint sequence;              //  请求编号
  191.     zmsg_t *request;            //  当前请求
  192.     zmsg_t *reply;              //  当前应答
  193.     int64_t expires;            //  请求过期时间
  194. } agent_t;
  196. agent_t *
  197. agent_new (zctx_t *ctx, void *pipe)
  198. {
  199.     agent_t *self = (agent_t *) zmalloc (sizeof (agent_t));
  200.     self->ctx = ctx;
  201.     self->pipe = pipe;
  202.     self->router = zsocket_new (self->ctx, ZMQ_ROUTER);
  203.     self->servers = zhash_new ();
  204.     self->actives = zlist_new ();
  205.     return self;
  206. }
  208. void
  209. agent_destroy (agent_t **self_p)
  210. {
  211.     assert (self_p);
  212.     if (*self_p) {
  213.         agent_t *self = *self_p;
  214.         zhash_destroy (&self->servers);
  215.         zlist_destroy (&self->actives);
  216.         zmsg_destroy (&self->request);
  217.         zmsg_destroy (&self->reply);
  218.         free (self);
  219.         *self_p = NULL;
  220.     }
  221. }
  223. //  当服务端从列表中移除时，回调该函数。
  225. static void
  226. s_server_free (void *argument)
  227. {
  228.     server_t *server = (server_t *) argument;
  229.     server_destroy (&server);
  230. }
  232. void
  233. agent_control_message (agent_t *self)
  234. {
  235.     zmsg_t *msg = zmsg_recv (self->pipe);
  236.     char *command = zmsg_popstr (msg);
  238.     if (streq (command, "CONNECT")) {
  239.         char *endpoint = zmsg_popstr (msg);
  240.         printf ("I: connecting to %s...\n", endpoint);
  241.         int rc = zmq_connect (self->router, endpoint);
  242.         assert (rc == 0);
  243.         server_t *server = server_new (endpoint);
  244.         zhash_insert (self->servers, endpoint, server);
  245.         zhash_freefn (self->servers, endpoint, s_server_free);
  246.         zlist_append (self->actives, server);
  247.         server->ping_at = zclock_time () + PING_INTERVAL;
  248.         server->expires = zclock_time () + SERVER_TTL;
  249.         free (endpoint);
  250.     }
  251.     else
  252.     if (streq (command, "REQUEST")) {
  253.         assert (!self->request);    //  遵循请求-应答循环
  254.         //  将请求编号和空帧加入消息顶部
  255.         char sequence_text [10];
  256.         sprintf (sequence_text, "%u", ++self->sequence);
  257.         zmsg_pushstr (msg, sequence_text);
  258.         //  获取请求消息的所有权
  259.         self->request = msg;
  260.         msg = NULL;
  261.         //  设置请求过期时间
  262.         self->expires = zclock_time () + GLOBAL_TIMEOUT;
  263.     }
  264.     free (command);
  265.     zmsg_destroy (&msg);
  266. }
  268. void
  269. agent_router_message (agent_t *self)
  270. {
  271.     zmsg_t *reply = zmsg_recv (self->router);
  273.     //  第一帧是应答的服务端标识
  274.     char *endpoint = zmsg_popstr (reply);
  275.     server_t *server =
  276.         (server_t *) zhash_lookup (self->servers, endpoint);
  277.     assert (server);
  278.     free (endpoint);
  279.     if (!server->alive) {
  280.         zlist_append (self->actives, server);
  281.         server->alive = 1;
  282.     }
  283.     server->ping_at = zclock_time () + PING_INTERVAL;
  284.     server->expires = zclock_time () + SERVER_TTL;
  286.     //  第二帧是应答的编号
  287.     char *sequence = zmsg_popstr (reply);
  288.     if (atoi (sequence) == self->sequence) {
  289.         zmsg_pushstr (reply, "OK");
  290.         zmsg_send (&reply, self->pipe);
  291.         zmsg_destroy (&self->request);
  292.     }
  293.     else
  294.         zmsg_destroy (&reply);
  295. }
  298. //  ---------------------------------------------------------------------
  299. //  异步的后台代理会维护一个服务端池，处理请求和应答。
  301. static void
  302. flcliapi_agent (void *args, zctx_t *ctx, void *pipe)
  303. {
  304.     agent_t *self = agent_new (ctx, pipe);
  306.     zmq_pollitem_t items [] = {
  307.         { self->pipe, 0, ZMQ_POLLIN, 0 },
  308.         { self->router, 0, ZMQ_POLLIN, 0 }
  309.     };
  310.     while (!zctx_interrupted) {
  311.         //  计算超时时间
  312.         uint64_t tickless = zclock_time () + 1000 * 3600;
  313.         if (self->request
  314.         &&  tickless > self->expires)
  315.             tickless = self->expires;
  316.         zhash_foreach (self->servers, server_tickless, &tickless);
  318.         int rc = zmq_poll (items, 2,
  319.             (tickless - zclock_time ()) * ZMQ_POLL_MSEC);
  320.         if (rc == -1)
  321.             break;              //  上下文对象被关闭
  323.         if (items [0].revents & ZMQ_POLLIN)
  324.             agent_control_message (self);
  326.         if (items [1].revents & ZMQ_POLLIN)
  327.             agent_router_message (self);
  329.         //  如果我们需要处理一项请求，将其发送给下一个可用的服务端
  330.         if (self->request) {
  331.             if (zclock_time () >= self->expires) {
  332.                 //  请求超时
  333.                 zstr_send (self->pipe, "FAILED");
  334.                 zmsg_destroy (&self->request);
  335.             }
  336.             else {
  337.                 //  寻找可用的服务端
  338.                 while (zlist_size (self->actives)) {
  339.                     server_t *server =
  340.                         (server_t *) zlist_first (self->actives);
  341.                     if (zclock_time () >= server->expires) {
  342.                         zlist_pop (self->actives);
  343.                         server->alive = 0;
  344.                     }
  345.                     else {
  346.                         zmsg_t *request = zmsg_dup (self->request);
  347.                         zmsg_pushstr (request, server->endpoint);
  348.                         zmsg_send (&request, self->router);
  349.                         break;
  350.                     }
  351.                 }
  352.             }
  353.         }
  354.         //  断开并删除已过期的服务端
  355.         //  发送心跳给空闲服务器
  356.         zhash_foreach (self->servers, server_ping, self->router);
  357.     }
  358.     agent_destroy (&self);
  359. }

这组API使用了较为复杂的机制，我们之前也有用到过：

异步后台代理

客户端API由两部分组成：同步的flcliapi类，运行于应用程序线程；异步的agent类，运行于后台线程。flcliapi和agent类通过一个inproc套接字互相通信。所有和ZMQ相关的内容都封装在API中。agent类实质上是作为一个迷你的代理程序在运行，负责在后台与服务端进行通信，只要我们发送请求，它就会设法连接一个服务器来处理请求。

连接等待机制

ROUTER套接字的特点之一是会直接丢弃无法路由的消息，这就意味着当与服务器建立了ROUTER-ROUTER连接后，如果立刻发送一条消息，该消息是会丢失的。flcliapi类则延迟了一会儿后再发送消息。之后的通信中，由于服务端套接字是持久的，客户端就不再丢弃消息了。

Ping silence

0MQ will queue messages for a dead server indefinitely. So if a client repeatedly PINGs a dead server, when that server comes back to life it’ll get a whole bunch of PING messages all at once. Rather than continuing to ping a server we know is offline, we count on 0MQ’s handling of durable sockets to deliver the old PING messages when the server comes back online. As soon as a server reconnects, it’ll get PINGs from all clients that were connected to it, it’ll PONG back, and those clients will recognize it as alive again.

调整轮询时间

在之前的示例程序中，我们一般会为轮询设置固定的超时时间（如1秒），这种做法虽然简单，但是对于用电较为敏感的设备来说（如笔记本电脑或手机）唤醒CPU是需要额外的电力的。所以，为了完美也好，好玩也好，我们这里调整了轮询时间，将其设置为到达过期时间时才超时，这样就能节省一部分轮询次数了。我们可以将过期时间放入一个列表中存储，方便查询。

总结

这一章中我们看到了很多可靠的请求-应答机制，每种机制都有其优劣性。大部分示例代码是可以直接用于生产环境的，不过还可以进一步优化。有两个模式会比较典型：使用了中间件的管家模式，以及未使用中间件的自由者模式。