zmq Pub-Sub 模式

5. Advanced Pub-Sub Patterns | ØMQ - The Guidehttps://zguide.zeromq.org/docs/chapter5/

我们将介绍：

何时使用发布订阅
如何处理太慢的订阅者（自杀蜗牛模式）
如何设计高速订阅者（黑盒模式）
如何监控发布-订阅网络（Espresso 模式）
如何构建共享键值存储（克隆模式）
如何使用反应器简化复杂的服务器
如何使用 Binary Star 模式向服务器添加故障转移

Pub-Sub 的优点和缺点
ZeroMQ 的低级模式有其不同的特点。 Pub-sub 解决了一个旧的消息传递问题，即多播或组消息传递。 ZeroMQ 将一丝不苟的简单性和残酷的冷漠结合在一起。值得了解 pub-sub 所做的权衡，这些权衡如何使我们受益，以及我们如何在需要时解决它们。

首先，PUB 将每条消息发送到“all of many”，而 PUSH 和 DEALER 将消息轮换为“one of many”。你不能简单地用 PUB 替换 PUSH，反之亦然，并希望事情会奏效。这值得重复，因为人们似乎经常建议这样做。

更深刻的是，pub-sub 的目标是可扩展性。这意味着大量数据会迅速发送给许多收件人。如果您需要每秒将数百万条消息发送到数千个点，那么与每秒向少数收件人发送几条消息相比，您会更喜欢发布订阅。

为了获得可扩展性，pub-sub 使用与推拉相同的技巧，即摆脱回声。这意味着收件人不会与发件人交谈。有一些例外，例如，SUB 套接字将向 PUB 套接字发送订阅，但它是匿名的且很少发生。

消除回声对于真正的可扩展性至关重要。使用 pub-sub，模式可以清晰地映射到 PGM 多播协议，该协议由网络交换机处理。换句话说，订阅者根本不连接到发布者，他们连接到交换机上的多播组，发布者将其消息发送到该组。

当我们消除回声时，我们的整体消息流变得更加简单，这让我们能够制作更简单的 API、更简单的协议，并且通常可以覆盖更多的人。但我们也排除了协调发送者和接收者的任何可能性。这意味着：

发布者无法判断订阅者何时成功连接，无论是在初始连接时还是在网络故障后重新连接时。
订阅者不能告诉发布者任何允许发布者控制他们发送消息速率的信息。发布者只有一种设置，即全速，订阅者要么跟上，要么丢失消息。
发布者无法判断订阅者何时因进程崩溃、网络中断等而消失。

缺点是如果我们想要进行可靠的多播，我们实际上需要所有这些。 ZeroMQ 发布-订阅模式会在订阅者连接时、网络故障发生时、或者订阅者或网络跟不上发布者时任意丢失消息。

好处是有很多用例几乎可靠的多播就好了。当我们需要这种回声时，我们可以切换到使用 ROUTER-DEALER（我倾向于在大多数正常音量情况下这样做），或者我们可以添加一个单独的同步通道（稍后我们将看到一个例子）本章）。

Pub-sub 就像无线电广播；加入之前你错过了一切，然后你得到多少信息取决于你的接待质量。令人惊讶的是，该模型非常有用且广泛，因为它完美地映射到现实世界的信息分布。想想 Facebook 和 Twitter、BBC World Service 和体育比赛结果。

正如我们对请求-回复所做的那样，让我们根据可能出现的问题来定义可靠性。以下是 pub-sub 的经典失败案例：

订阅者加入晚了，所以他们错过了服务器已经发送的消息。
订阅者获取消息的速度可能太慢，因此队列会建立起来然后溢出。
订阅者可能会在离开时掉线并丢失消息。
订阅者可能会崩溃并重新启动，并丢失他们已经收到的任何数据。
网络可能会过载并丢失数据（特别是对于 PGM）。
网络可能会变得太慢，因此发布者端队列溢出并导致发布者崩溃。

更多可能出错，但这些是我们在现实系统中看到的典型故障。从 v3.x 开始，ZeroMQ 对其内部缓冲区（所谓的高水位标记或 HWM）强制默认限制，因此除非您故意将 HWM 设置为无限，否则发布者崩溃的情况很少见。

所有这些失败案例都有答案，尽管并不总是简单的。可靠性需要我们大多数人不需要的复杂性，大多数时候，这就是 ZeroMQ 不尝试开箱即用提供它的原因（即使有一个全局可靠性设计，但没有）。

Pub-Sub 跟踪（浓缩咖啡模式）#
让我们通过研究一种跟踪发布-订阅网络的方法来开始本章。在第 2 章 - 套接字和模式中，我们看到了一个简单的代理，它使用这些来进行传输桥接。 zmq_proxy() 方法有三个参数：一个桥接在一起的前端和后端套接字，以及一个将所有消息发送到的捕获套接字。

代码看似简单：

//  Espresso Pattern
//  This shows how to capture data using a pub-sub proxy#include "czmq.h"//  The subscriber thread requests messages starting with
//  A and B, then reads and counts incoming messages.static void
subscriber_thread (void *args, zctx_t *ctx, void *pipe)
{//  Subscribe to "A" and "B"void *subscriber = zsocket_new (ctx, ZMQ_SUB);zsocket_connect (subscriber, "tcp://localhost:6001");zsocket_set_subscribe (subscriber, "A");zsocket_set_subscribe (subscriber, "B");int count = 0;while (count < 5) {char *string = zstr_recv (subscriber);if (!string)break;              //  Interruptedfree (string);count++;}zsocket_destroy (ctx, subscriber);
}//  .split publisher thread
//  The publisher sends random messages starting with A-J:static void
publisher_thread (void *args, zctx_t *ctx, void *pipe)
{void *publisher = zsocket_new (ctx, ZMQ_PUB);zsocket_bind (publisher, "tcp://*:6000");while (!zctx_interrupted) {char string [10];sprintf (string, "%c-%05d", randof (10) + 'A', randof (100000));if (zstr_send (publisher, string) == -1)break;              //  Interruptedzclock_sleep (100);     //  Wait for 1/10th second}
}//  .split listener thread
//  The listener receives all messages flowing through the proxy, on its
//  pipe. In CZMQ, the pipe is a pair of ZMQ_PAIR sockets that connect
//  attached child threads. In other languages your mileage may vary:static void
listener_thread (void *args, zctx_t *ctx, void *pipe)
{//  Print everything that arrives on pipewhile (true) {zframe_t *frame = zframe_recv (pipe);if (!frame)break;              //  Interruptedzframe_print (frame, NULL);zframe_destroy (&frame);}
}//  .split main thread
//  The main task starts the subscriber and publisher, and then sets
//  itself up as a listening proxy. The listener runs as a child thread:int main (void)
{//  Start child threadszctx_t *ctx = zctx_new ();zthread_fork (ctx, publisher_thread, NULL);zthread_fork (ctx, subscriber_thread, NULL);void *subscriber = zsocket_new (ctx, ZMQ_XSUB);zsocket_connect (subscriber, "tcp://localhost:6000");void *publisher = zsocket_new (ctx, ZMQ_XPUB);zsocket_bind (publisher, "tcp://*:6001");void *listener = zthread_fork (ctx, listener_thread, NULL);zmq_proxy (subscriber, publisher, listener);puts (" interrupted");//  Tell attached threads to exitzctx_destroy (&ctx);return 0;
}

Espresso 通过创建一个侦听器线程来工作，该线程读取 PAIR 套接字并打印它获得的任何内容。 PAIR 套接字是管道的一端；另一端（另一个 PAIR）是我们传递给 zmq_proxy() 的套接字。在实践中，您可以过滤有趣的消息以获得想要跟踪的内容的本质（因此模式的名称）。

订阅者线程订阅“A”和“B”，收到五条消息，然后销毁它的套接字。运行示例时，侦听器打印两条订阅消息、五条数据消息、两条取消订阅消息，然后静默：

[002] 0141
[002] 0142
[007] B-91164
[007] B-12979
[007] A-52599
[007] A-06417
[007] A-45770
[002] 0041
[002] 0042

这清楚地显示了发布者套接字在没有订阅者时如何停止发送数据。发布者线程仍在发送消息。套接字只是默默地放下它们。

上次值缓存
如果您使用过商业发布-订阅系统，您可能已经习惯了快速而愉快的 ZeroMQ 发布-订阅模型中缺少的一些功能。其中之一是最后一个值缓存（LVC）。这解决了新用户加入网络时如何赶上的问题。该理论是，当新订阅者加入并订阅某些特定主题时，发布者会收到通知。然后发布者可以为这些主题重新广播最后一条消息。

我已经解释了为什么当有新订阅者时发布者不会得到通知，因为在大型发布订阅系统中，数据量使其几乎不可能。要构建真正大规模的发布-订阅网络，您需要一个类似于 PGM 的协议，该协议利用高端以太网交换机向数千个订阅者多播数据的能力。尝试从发布者到数千个订阅者中的每一个都进行 TCP 单播是无法扩展的。你会遇到奇怪的峰值、不公平的分配（一些订阅者比其他订阅者更早地收到消息）、网络拥塞和普遍的不满。

PGM 是一种单向协议：发布者将消息发送到交换机的多播地址，然后将其重新广播给所有感兴趣的订阅者。发布者永远不会看到订阅者何时加入或离开：这一切都发生在交换机中，我们真的不想开始重新编程。

然而，在拥有几十个订阅者和有限数量主题的低容量网络中，我们可以使用 TCP，然后 XSUB 和 XPUB 套接字确实可以相互通信，就像我们刚刚在 Espresso 模式中看到的那样。

我们可以使用 ZeroMQ 制作 LVC 吗？答案是肯定的，如果我们在发布者和订阅者之间建立一个代理； PGM 开关的模拟，但我们可以自己编程。

我将首先创建一个突出显示最坏情况的发布者和订阅者。这个出版商是病态的。它首先立即向一千个主题中的每一个发送消息，然后每秒向随机主题发送一个更新。订阅者连接并订阅主题。如果没有 LVC，订阅者必须平均等待 500 秒才能获取任何数据。为了增加一些戏剧性，让我们假设有一个名叫 Gregor 的逃犯威胁说，如果我们不能解决 8.3 分钟的延迟，就要把玩具兔子 Roger 的头扯下来。

这是发布者代码。请注意，它具有连接到某个地址的命令行选项，但以其他方式绑定到端点。我们稍后将使用它来连接到我们的最后一个值缓存：

//  Pathological publisher
//  Sends out 1,000 topics and then one random update per second#include "czmq.h"int main (int argc, char *argv [])
{zctx_t *context = zctx_new ();void *publisher = zsocket_new (context, ZMQ_PUB);if (argc == 2)zsocket_bind (publisher, argv [1]);elsezsocket_bind (publisher, "tcp://*:5556");//  Ensure subscriber connection has time to completesleep (1);//  Send out all 1,000 topic messagesint topic_nbr;for (topic_nbr = 0; topic_nbr < 1000; topic_nbr++) {zstr_sendfm (publisher, "%03d", topic_nbr);zstr_send (publisher, "Save Roger");}//  Send one random update per secondsrandom ((unsigned) time (NULL));while (!zctx_interrupted) {sleep (1);zstr_sendfm (publisher, "%03d", randof (1000));zstr_send (publisher, "Off with his head!");}zctx_destroy (&context);return 0;
}

这是订阅者:

//  Subscribes to one random topic and prints received messages#include "czmq.h"int main (int argc, char *argv [])
{zctx_t *context = zctx_new ();void *subscriber = zsocket_new (context, ZMQ_SUB);if (argc == 2)zsocket_connect (subscriber, argv [1]);elsezsocket_connect (subscriber, "tcp://localhost:5556");srandom ((unsigned) time (NULL));char subscription [5];sprintf (subscription, "%03d", randof (1000));zsocket_set_subscribe (subscriber, subscription);while (true) {char *topic = zstr_recv (subscriber);if (!topic)break;char *data = zstr_recv (subscriber);assert (streq (topic, subscription));puts (data);free (topic);free (data);}zctx_destroy (&context);return 0;
}

尝试构建和运行这些：首先是订阅者，然后是发布者。您会看到订阅者报告如您所愿地显示“Save Roger”:

./pathosub &
./pathopub

当您运行第二个订阅者时，您就会了解罗杰的困境。在它报告获取任何数据之前，您必须将其放置很长时间。所以，这是我们的最后一个值缓存。正如我所承诺的，它是一个绑定到两个套接字然后在两个套接字上处理消息的代理:

//  Last value cache
//  Uses XPUB subscription messages to re-send data#include "czmq.h"int main (void)
{zctx_t *context = zctx_new ();void *frontend = zsocket_new (context, ZMQ_SUB);zsocket_connect (frontend, "tcp://*:5557");void *backend = zsocket_new (context, ZMQ_XPUB);zsocket_bind (backend, "tcp://*:5558");//  Subscribe to every single topic from publisherzsocket_set_subscribe (frontend, "");//  Store last instance of each topic in a cachezhash_t *cache = zhash_new ();//  .split main poll loop//  We route topic updates from frontend to backend, and//  we handle subscriptions by sending whatever we cached,//  if anything:while (true) {zmq_pollitem_t items [] = {{ frontend, 0, ZMQ_POLLIN, 0 },{ backend,  0, ZMQ_POLLIN, 0 }};if (zmq_poll (items, 2, 1000 * ZMQ_POLL_MSEC) == -1)break;              //  Interrupted//  Any new topic data we cache and then forwardif (items [0].revents & ZMQ_POLLIN) {char *topic = zstr_recv (frontend);char *current = zstr_recv (frontend);if (!topic)break;char *previous = zhash_lookup (cache, topic);if (previous) {zhash_delete (cache, topic);free (previous);}zhash_insert (cache, topic, current);zstr_sendm (backend, topic);zstr_send (backend, current);free (topic);}//  .split handle subscriptions//  When we get a new subscription, we pull data from the cache:if (items [1].revents & ZMQ_POLLIN) {zframe_t *frame = zframe_recv (backend);if (!frame)break;//  Event is one byte 0=unsub or 1=sub, followed by topicbyte *event = zframe_data (frame);if (event [0] == 1) {char *topic = zmalloc (zframe_size (frame));memcpy (topic, event + 1, zframe_size (frame) - 1);printf ("Sending cached topic %s\n", topic);char *previous = zhash_lookup (cache, topic);if (previous) {zstr_sendm (backend, topic);zstr_send (backend, previous);}free (topic);}zframe_destroy (&frame);}}zctx_destroy (&context);zhash_destroy (&cache);return 0;
}

现在，运行代理，然后是发布者：

./lvcache &
./pathopub tcp://localhost:5557

现在运行尽可能多的订阅者实例，每次连接到端口 5558 上的代理：

./pathosub tcp://localhost:5558

每个订阅者都高兴地报告“拯救罗杰”，逃犯格雷戈尔偷偷溜回座位吃晚饭和一杯好喝的热牛奶，这是他最初真正想要的。

一个注意事项：默认情况下，XPUB 套接字不会报告重复订阅，这正是您天真地将 XPUB 连接到 XSUB 时所需要的。我们的示例通过使用随机主题偷偷摸摸地解决了这个问题，因此它不起作用的可能性是百万分之一。在真正的 LVC 代理中，您需要使用我们在第 6 章 - ZeroMQ 社区中实现的 ZMQ_XPUB_VERBOSE 选项作为练习

慢速订阅者检测（自杀蜗牛模式）#
在现实生活中使用 pub-sub 模式时会遇到的一个常见问题是订阅者缓慢。在理想的世界中，我们以全速将数据从发布者传输到订阅者。实际上，订阅者应用程序通常是用解释性语言编写的，或者只是做了很多工作，或者只是写得不好，以至于它们跟不上发布者的步伐。

我们如何处理缓慢的订阅者？理想的解决方法是让订阅者更快，但这可能需要工作和时间。处理慢速订阅者的一些经典策略是：

在发布者上排队消息。当我几个小时不阅读电子邮件时，Gmail 就是这样做的。但是在大容量消息传递中，将队列推向上游会导致发布者耗尽内存并崩溃——尤其是当订阅者很多并且出于性能原因无法刷新到磁盘时，会产生令人兴奋但无利可图的结果。
在订阅者上排队消息。这要好得多，如果网络可以跟上，这就是 ZeroMQ 默认情况下所做的。如果有人要耗尽内存并崩溃，那将是订阅者而不是发布者，这是公平的。这对于“峰值”流来说是完美的，在这种情况下，订阅者无法跟上一段时间，但在流变慢时可以赶上。但是，对于一般来说太慢的订阅者来说，这不是答案。
一段时间后停止排队新消息。这就是当我的邮箱溢出其宝贵的 GB 空间时 Gmail 所做的事情。新消息只会被拒绝或丢弃。从发布者的角度来看，这是一个很好的策略，这也是发布者设置 HWM 时 ZeroMQ 所做的。但是，它仍然无法帮助我们修复缓慢的订阅者。现在我们只是在我们的消息流中出现了间隙。
用断开连接惩罚慢速订阅者。这就是 Hotmail（还记得吗？）在我两周没有登录时所做的事情，这就是为什么当我发现可能有更好的方法时，我正在使用第 15 个 Hotmail 帐户。这是一个很好的残酷策略，它迫使订阅者坐起来集中注意力，这将是理想的，但 ZeroMQ 没有这样做，并且无法将其分层，因为订阅者对于发布者应用程序是不可见的。

这些经典策略都不适合，所以我们需要发挥创意。与其断开发布者的连接，不如说服订阅者自杀。这是自杀蜗牛模式。当订阅者检测到它运行得太慢时（其中“太慢”大概是一个配置选项，真正的意思是“太慢了，如果你到达这里，请大声喊叫，因为我需要知道，所以我可以解决这个问题！”），它会发出嘶嘶声并死去。

订户如何检测到这一点？一种方法是对消息进行排序（按顺序编号）并在发布者处使用 HWM。现在，如果订阅者检测到一个间隙（即编号不连续），它就知道有问题。然后我们将 HWM 调整到“如果你达到这个水平就会发出嘶嘶声和死亡”的水平。

这个解决方案有两个问题。一，如果我们有很多发布者，我们如何对消息进行排序？解决方案是给每个发布者一个唯一的 ID 并将其添加到排序中。其次，如果订阅者使用 ZMQ_SUBSCRIBE 过滤器，他们将根据定义获得间隙。我们宝贵的测序将一事无成。

一些用例不会使用过滤器，排序将适用于它们。但更通用的解决方案是发布者为每条消息添加时间戳。当订阅者收到一条消息时，它会检查时间，如果差异超过一秒，它就会执行“吱吱作响”的事情，可能首先向某个操作员控制台发出尖叫声。

Suicide Snail 模式尤其适用于订阅者拥有自己的客户和服务级别协议并且需要保证某些最大延迟的情况。中止订阅者似乎不是保证最大延迟的建设性方法，但它是断言模型。今天中止，问题将得到解决。允许迟到的数据向下游流动，问题可能会造成更大范围的损害，并且需要更长的时间才会出现在雷达上。

这是一个自杀蜗牛的最小例子：

//  Suicidal Snail#include "czmq.h"//  This is our subscriber. It connects to the publisher and subscribes
//  to everything. It sleeps for a short time between messages to
//  simulate doing too much work. If a message is more than one second
//  late, it croaks.#define MAX_ALLOWED_DELAY   1000    //  msecsstatic void
subscriber (void *args, zctx_t *ctx, void *pipe)
{//  Subscribe to everythingvoid *subscriber = zsocket_new (ctx, ZMQ_SUB);zsocket_set_subscribe (subscriber, "");zsocket_connect (subscriber, "tcp://localhost:5556");//  Get and process messageswhile (true) {char *string = zstr_recv (subscriber);printf("%s\n", string);int64_t clock;int terms = sscanf (string, "%" PRId64, &clock);assert (terms == 1);free (string);//  Suicide snail logicif (zclock_time () - clock > MAX_ALLOWED_DELAY) {fprintf (stderr, "E: subscriber cannot keep up, aborting\n");break;}//  Work for 1 msec plus some random additional timezclock_sleep (1 + randof (2));}zstr_send (pipe, "gone and died");
}//  .split publisher task
//  This is our publisher task. It publishes a time-stamped message to its
//  PUB socket every millisecond:static void
publisher (void *args, zctx_t *ctx, void *pipe)
{//  Prepare publishervoid *publisher = zsocket_new (ctx, ZMQ_PUB);zsocket_bind (publisher, "tcp://*:5556");while (true) {//  Send current clock (msecs) to subscriberschar string [20];sprintf (string, "%" PRId64, zclock_time ());zstr_send (publisher, string);char *signal = zstr_recv_nowait (pipe);if (signal) {free (signal);break;}zclock_sleep (1);            //  1msec wait}
}//  .split main task
//  The main task simply starts a client and a server, and then
//  waits for the client to signal that it has died:int main (void)
{zctx_t *ctx = zctx_new ();void *pubpipe = zthread_fork (ctx, publisher, NULL);void *subpipe = zthread_fork (ctx, subscriber, NULL);free (zstr_recv (subpipe));zstr_send (pubpipe, "break");zclock_sleep (100);zctx_destroy (&ctx);return 0;
}

以下是有关“自杀蜗牛”示例的一些注意事项：

此处的消息仅包含以毫秒为单位的当前系统时钟。在实际的应用程序中，您至少有一个带有时间戳的消息头和一个带有数据的消息体。
该示例在单个进程中将订阅者和发布者作为两个线程。实际上，它们将是单独的过程。使用线程只是为了演示方便。

高速用户（黑盒模式）
现在让我们看看一种让我们的订阅者更快的方法。 pub-sub 的一个常见用例是分发大型数据流，例如来自证券交易所的市场数据。典型的设置是让发布者连接到证券交易所，获取报价并将其发送给许多订阅者。如果有少量订阅者，我们可以使用 TCP。如果我们有更多的订阅者，我们可能会使用可靠的多播，即 PGM。

图 56 - 简单的黑盒模式

假设我们的提要每秒平均有 100,000 条 100 字节的消息。这是一个典型的费率，在过滤市场数据后，我们不需要发送给订阅者。现在我们决定记录一天的数据（8 小时内可能有 250 GB），然后将其重播到模拟网络，即一小群订阅者。虽然对于 ZeroMQ 应用程序来说每秒 100K 条消息很容易，但我们希望重播它的速度要快得多。

所以我们用一堆盒子建立了我们的架构——一个用于发布者，一个用于每个订阅者。这些是明确指定的盒子——八个核心，十二个给出版商。

当我们将数据注入订阅者时，我们注意到两件事：

当我们对一条消息做哪怕是最轻微的工作时，它都会使我们的订阅者减慢到无法再次赶上发布者的程度。
我们在发布者和订阅者处达到了每秒 600 万条消息的上限，即使经过仔细优化和 TCP 调整。

我们要做的第一件事是将订阅者分成多线程设计，这样我们就可以在一组线程中处理消息，同时在另一组线程中读取消息。通常，我们不希望以相同的方式处理每条消息。相反，订阅者将过滤一些消息，可能是通过前缀键。当消息符合某些条件时，订阅者将调用一个工作程序来处理它。在 ZeroMQ 术语中，这意味着将消息发送到工作线程。

所以订阅者看起来像一个队列设备。我们可以使用各种套接字来连接订阅者和工作人员。如果我们假设单向流量和工作线程都是相同的，我们可以使用 PUSH 和 PULL 并将所有路由工作委托给 ZeroMQ。这是最简单、最快的方法。

订阅者通过 TCP 或 PGM 与发布者交谈。订阅者通过 inproc:@<//>@ 与其在同一个进程中的工作人员交谈。

图 57 - 疯狂的黑盒模式

现在打破那个天花板。订阅者线程占用了 100% 的 CPU，因为它是一个线程，所以不能使用多个内核。单个线程总是会达到上限，无论是每秒 2M、6M 还是更多消息。我们希望将工作拆分到可以并行运行的多个线程中。

许多高性能产品使用的方法在这里有效，就是分片。使用分片，我们将工作分成并行和独立的流，例如一个流中的一半主题键，另一个流中的一半。我们可以使用许多流，但除非我们有免费的内核，否则性能将无法扩展。那么让我们看看如何分片成两个流。

有两个流，全速工作，我们将配置 ZeroMQ 如下：

两个 I/O 线程，而不是一个。
两个网络接口 (NIC)，每个订阅者一个。
每个 I/O 线程绑定到特定的 NIC。
两个订阅者线程，绑定到特定内核。
两个 SUB 套接字，每个订阅者线程一个。
剩余的内核分配给工作线程。
工作线程连接到两个订阅者 PUSH 套接字。

理想情况下，我们希望将架构中满载线程的数量与内核数量相匹配。当线程开始争夺内核和 CPU 周期时，添加更多线程的成本超过了收益。例如，创建更多 I/O 线程将没有任何好处。

可靠的发布订阅（克隆模式）
作为一个更大的工作示例，我们将讨论制作可靠的发布-订阅架构的问题。我们将分阶段开发。目标是允许一组应用程序共享一些公共状态。以下是我们的技术挑战：

我们有大量的客户端应用程序，比如说数千或数万。
他们会随意加入和离开网络。
这些应用程序必须共享一个最终一致的状态。
任何应用程序都可以在任何时间点更新状态。

假设更新量相当低。我们没有实时目标。整个状态可以放入内存中。一些合理的用例是：

一组云服务器共享的配置。
一组玩家共享的一些游戏状态。
实时更新并可供应用程序使用的汇率数据。

中心化与去中心化
我们必须做出的第一个决定是我们是否使用中央服务器。它对最终的设计产生了很大的影响。权衡是这些：

从概念上讲，中央服务器更容易理解，因为网络不是自然对称的。使用中央服务器，我们可以避免发现、绑定与连接等所有问题。
通常，完全分布式架构在技术上更具挑战性，但最终会得到更简单的协议。也就是说，每个节点都必须以正确的方式充当服务器和客户端，这很微妙。如果做得好，结果比使用中央服务器更简单。我们在第 4 章 - 可靠的请求-回复模式的自由职业者模式中看到了这一点。
中央服务器将成为大容量用例的瓶颈。如果需要每秒处理数百万条消息的规模，我们应该立即以去中心化为目标。
具有讽刺意味的是，集中式架构比分散式架构更容易扩展到更多节点。也就是说，将 10,000 个节点连接到一台服务器比相互连接更容易。

因此，对于克隆模式，我们将使用发布状态更新的服务器和代表应用程序的一组客户端。

将状态表示为键值对 #
我们将分阶段开发 Clone，一次解决一个问题。首先，让我们看看如何在一组客户端之间更新共享状态。我们需要决定如何表示我们的状态，以及更新。最简单的合理格式是键值存储，其中一个键值对表示共享状态中的一个原子变化单元。

我们在第 1 章 - 基础知识、天气服务器和客户端中有一个简单的发布-订阅示例。让我们更改服务器以发送键值对，而客户端将它们存储在哈希表中。这让我们可以使用经典的 pub-sub 模型将更新从一个服务器发送到一组客户端。

更新是新的键值对、现有键的修改值或删除的键。我们现在可以假设整个存储都适合内存，并且应用程序通过键访问它，例如使用哈希表或字典。对于更大的存储和某种持久性，我们可能会将状态存储在数据库中，但这在这里无关紧要。

这是服务器：

//  Clone server Model One#include "kvsimple.c"int main (void)
{//  Prepare our context and publisher socketzctx_t *ctx = zctx_new ();void *publisher = zsocket_new (ctx, ZMQ_PUB);zsocket_bind (publisher, "tcp://*:5556");zclock_sleep (200);zhash_t *kvmap = zhash_new ();int64_t sequence = 0;srandom ((unsigned) time (NULL));while (!zctx_interrupted) {//  Distribute as key-value messagekvmsg_t *kvmsg = kvmsg_new (++sequence);kvmsg_fmt_key  (kvmsg, "%d", randof (10000));kvmsg_fmt_body (kvmsg, "%d", randof (1000000));kvmsg_send     (kvmsg, publisher);kvmsg_store   (&kvmsg, kvmap);}printf (" Interrupted\n%d messages out\n", (int) sequence);zhash_destroy (&kvmap);zctx_destroy (&ctx);return 0;
}

这是客户端：

//  Clone client Model One#include "kvsimple.c"int main (void)
{//  Prepare our context and updates socketzctx_t *ctx = zctx_new ();void *updates = zsocket_new (ctx, ZMQ_SUB);zsocket_set_subscribe (updates, "");zsocket_connect (updates, "tcp://localhost:5556");zhash_t *kvmap = zhash_new ();int64_t sequence = 0;while (true) {kvmsg_t *kvmsg = kvmsg_recv (updates);if (!kvmsg)break;          //  Interruptedkvmsg_store (&kvmsg, kvmap);sequence++;}printf (" Interrupted\n%d messages in\n", (int) sequence);zhash_destroy (&kvmap);zctx_destroy (&ctx);return 0;
}

图 58 - 发布状态更新

以下是关于第一个模型的一些注意事项：

所有艰苦的工作都在 kvmsg 类中完成。此类使用键值消息对象，这些对象是由三个框架构成的多部分 ZeroMQ 消息：一个键（一个 ZeroMQ 字符串）、一个序列号（64 位值，按网络字节顺序）和一个二进制主体（包含所有内容）别的）。
服务器使用随机的 4 位密钥生成消息，这让我们可以模拟一个很大但不是很大的哈希表（10K 条目）。
我们在此版本中不实现删除：所有消息都是插入或更新。
服务器在绑定套接字后会暂停 200 毫秒。这是为了防止慢连接综合症，即订阅者在连接到服务器的套接字时丢失消息。我们将在更高版本的克隆代码中删除它。
我们将在代码中使用术语发布者和订阅者来指代套接字。当我们有多个套接字做不同的事情时，这将有所帮助。

这是 kvmsg 类，目前最简单的形式：

//  kvsimple class - key-value message class for example applications#include "kvsimple.h"
#include "zlist.h"//  Keys are short strings
#define KVMSG_KEY_MAX   255//  Message is formatted on wire as 3 frames:
//  frame 0: key (0MQ string)
//  frame 1: sequence (8 bytes, network order)
//  frame 2: body (blob)
#define FRAME_KEY       0
#define FRAME_SEQ       1
#define FRAME_BODY      2
#define KVMSG_FRAMES    3//  The kvmsg class holds a single key-value message consisting of a
//  list of 0 or more frames:struct _kvmsg {//  Presence indicators for each frameint present [KVMSG_FRAMES];//  Corresponding 0MQ message frames, if anyzmq_msg_t frame [KVMSG_FRAMES];//  Key, copied into safe C stringchar key [KVMSG_KEY_MAX + 1];
};//  .split constructor and destructor
//  Here are the constructor and destructor for the class://  Constructor, takes a sequence number for the new kvmsg instance:
kvmsg_t *
kvmsg_new (int64_t sequence)
{kvmsg_t*self;self = (kvmsg_t *) zmalloc (sizeof (kvmsg_t));kvmsg_set_sequence (self, sequence);return self;
}//  zhash_free_fn callback helper that does the low level destruction:
void
kvmsg_free (void *ptr)
{if (ptr) {kvmsg_t *self = (kvmsg_t *) ptr;//  Destroy message frames if anyint frame_nbr;for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++)if (self->present [frame_nbr])zmq_msg_close (&self->frame [frame_nbr]);//  Free object itselffree (self);}
}//  Destructor
void
kvmsg_destroy (kvmsg_t **self_p)
{assert (self_p);if (*self_p) {kvmsg_free (*self_p);*self_p = NULL;}
}//  .split recv method
//  This method reads a key-value message from socket, and returns a new
//  {{kvmsg}} instance:kvmsg_t *
kvmsg_recv (void *socket)
{assert (socket);kvmsg_t *self = kvmsg_new (0);//  Read all frames off the wire, reject if bogusint frame_nbr;for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++) {if (self->present [frame_nbr])zmq_msg_close (&self->frame [frame_nbr]);zmq_msg_init (&self->frame [frame_nbr]);self->present [frame_nbr] = 1;if (zmq_msg_recv (&self->frame [frame_nbr], socket, 0) == -1) {kvmsg_destroy (&self);break;}//  Verify multipart framingint rcvmore = (frame_nbr < KVMSG_FRAMES - 1)? 1: 0;if (zsocket_rcvmore (socket) != rcvmore) {kvmsg_destroy (&self);break;}}return self;
}//  .split send method
//  This method sends a multiframe key-value message to a socket:void
kvmsg_send (kvmsg_t *self, void *socket)
{assert (self);assert (socket);int frame_nbr;for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++) {zmq_msg_t copy;zmq_msg_init (&copy);if (self->present [frame_nbr])zmq_msg_copy (&copy, &self->frame [frame_nbr]);zmq_msg_send (&copy, socket, (frame_nbr < KVMSG_FRAMES - 1)? ZMQ_SNDMORE: 0);zmq_msg_close (&copy);}
}//  .split key methods
//  These methods let the caller get and set the message key, as a
//  fixed string and as a printf formatted string:char *
kvmsg_key (kvmsg_t *self)
{assert (self);if (self->present [FRAME_KEY]) {if (!*self->key) {size_t size = zmq_msg_size (&self->frame [FRAME_KEY]);if (size > KVMSG_KEY_MAX)size = KVMSG_KEY_MAX;memcpy (self->key,zmq_msg_data (&self->frame [FRAME_KEY]), size);self->key [size] = 0;}return self->key;}elsereturn NULL;
}void
kvmsg_set_key (kvmsg_t *self, char *key)
{assert (self);zmq_msg_t *msg = &self->frame [FRAME_KEY];if (self->present [FRAME_KEY])zmq_msg_close (msg);zmq_msg_init_size (msg, strlen (key));memcpy (zmq_msg_data (msg), key, strlen (key));self->present [FRAME_KEY] = 1;
}void
kvmsg_fmt_key (kvmsg_t *self, char *format, ...)
{char value [KVMSG_KEY_MAX + 1];va_list args;assert (self);va_start (args, format);vsnprintf (value, KVMSG_KEY_MAX, format, args);va_end (args);kvmsg_set_key (self, value);
}//  .split sequence methods
//  These two methods let the caller get and set the message sequence number:int64_t
kvmsg_sequence (kvmsg_t *self)
{assert (self);if (self->present [FRAME_SEQ]) {assert (zmq_msg_size (&self->frame [FRAME_SEQ]) == 8);byte *source = zmq_msg_data (&self->frame [FRAME_SEQ]);int64_t sequence = ((int64_t) (source [0]) << 56)+ ((int64_t) (source [1]) << 48)+ ((int64_t) (source [2]) << 40)+ ((int64_t) (source [3]) << 32)+ ((int64_t) (source [4]) << 24)+ ((int64_t) (source [5]) << 16)+ ((int64_t) (source [6]) << 8)+  (int64_t) (source [7]);return sequence;}elsereturn 0;
}void
kvmsg_set_sequence (kvmsg_t *self, int64_t sequence)
{assert (self);zmq_msg_t *msg = &self->frame [FRAME_SEQ];if (self->present [FRAME_SEQ])zmq_msg_close (msg);zmq_msg_init_size (msg, 8);byte *source = zmq_msg_data (msg);source [0] = (byte) ((sequence >> 56) & 255);source [1] = (byte) ((sequence >> 48) & 255);source [2] = (byte) ((sequence >> 40) & 255);source [3] = (byte) ((sequence >> 32) & 255);source [4] = (byte) ((sequence >> 24) & 255);source [5] = (byte) ((sequence >> 16) & 255);source [6] = (byte) ((sequence >> 8)  & 255);source [7] = (byte) ((sequence)       & 255);self->present [FRAME_SEQ] = 1;
}//  .split message body methods
//  These methods let the caller get and set the message body as a
//  fixed string and as a printf formatted string:byte *
kvmsg_body (kvmsg_t *self)
{assert (self);if (self->present [FRAME_BODY])return (byte *) zmq_msg_data (&self->frame [FRAME_BODY]);elsereturn NULL;
}void
kvmsg_set_body (kvmsg_t *self, byte *body, size_t size)
{assert (self);zmq_msg_t *msg = &self->frame [FRAME_BODY];if (self->present [FRAME_BODY])zmq_msg_close (msg);self->present [FRAME_BODY] = 1;zmq_msg_init_size (msg, size);memcpy (zmq_msg_data (msg), body, size);
}void
kvmsg_fmt_body (kvmsg_t *self, char *format, ...)
{char value [255 + 1];va_list args;assert (self);va_start (args, format);vsnprintf (value, 255, format, args);va_end (args);kvmsg_set_body (self, (byte *) value, strlen (value));
}//  .split size method
//  This method returns the body size of the most recently read message,
//  if any exists:size_t
kvmsg_size (kvmsg_t *self)
{assert (self);if (self->present [FRAME_BODY])return zmq_msg_size (&self->frame [FRAME_BODY]);elsereturn 0;
}//  .split store method
//  This method stores the key-value message into a hash map, unless
//  the key and value are both null. It nullifies the {{kvmsg}} reference
//  so that the object is owned by the hash map, not the caller:void
kvmsg_store (kvmsg_t **self_p, zhash_t *hash)
{assert (self_p);if (*self_p) {kvmsg_t *self = *self_p;assert (self);if (self->present [FRAME_KEY]&&  self->present [FRAME_BODY]) {zhash_update (hash, kvmsg_key (self), self);zhash_freefn (hash, kvmsg_key (self), kvmsg_free);}*self_p = NULL;}
}//  .split dump method
//  This method prints the key-value message to stderr for
//  debugging and tracing:void
kvmsg_dump (kvmsg_t *self)
{if (self) {if (!self) {fprintf (stderr, "NULL");return;}size_t size = kvmsg_size (self);byte  *body = kvmsg_body (self);fprintf (stderr, "[seq:%" PRId64 "]", kvmsg_sequence (self));fprintf (stderr, "[key:%s]", kvmsg_key (self));fprintf (stderr, "[size:%zd] ", size);int char_nbr;for (char_nbr = 0; char_nbr < size; char_nbr++)fprintf (stderr, "%02X", body [char_nbr]);fprintf (stderr, "\n");}elsefprintf (stderr, "NULL message\n");
}//  .split test method
//  It's good practice to have a self-test method that tests the class; this
//  also shows how it's used in applications:int
kvmsg_test (int verbose)
{kvmsg_t*kvmsg;printf (" * kvmsg: ");//  Prepare our context and socketszctx_t *ctx = zctx_new ();void *output = zsocket_new (ctx, ZMQ_DEALER);int rc = zmq_bind (output, "ipc://kvmsg_selftest.ipc");assert (rc == 0);void *input = zsocket_new (ctx, ZMQ_DEALER);rc = zmq_connect (input, "ipc://kvmsg_selftest.ipc");assert (rc == 0);zhash_t *kvmap = zhash_new ();//  Test send and receive of simple messagekvmsg = kvmsg_new (1);kvmsg_set_key  (kvmsg, "key");kvmsg_set_body (kvmsg, (byte *) "body", 4);if (verbose)kvmsg_dump (kvmsg);kvmsg_send (kvmsg, output);kvmsg_store (&kvmsg, kvmap);kvmsg = kvmsg_recv (input);if (verbose)kvmsg_dump (kvmsg);assert (streq (kvmsg_key (kvmsg), "key"));kvmsg_store (&kvmsg, kvmap);//  Shutdown and destroy all objectszhash_destroy (&kvmap);zctx_destroy (&ctx);printf ("OK\n");return 0;
}

稍后，我们将创建一个更复杂的 kvmsg 类，它可以在实际应用中使用。

服务器和客户端都维护哈希表，但只有当我们在服务器之前启动所有客户端并且客户端永远不会崩溃时，第一个模型才能正常工作。这是非常人为的。

获取带外快照 #
所以现在我们有第二个问题：如何处理迟加入的客户端或崩溃然后重新启动的客户端。

为了让迟到（或正在恢复）的客户端赶上服务器，它必须获取服务器状态的快照。正如我们将“消息”简化为“一个有序的键值对”一样，我们也可以将“状态”简化为“一个哈希表”。为了获得服务器状态，客户端打开一个 DEALER 套接字并显式地请求它。

为了使这项工作发挥作用，我们必须解决时间问题。获取状态快照需要一定的时间，如果快照很大，可能需要相当长的时间。我们需要正确地将更新应用到快照。但是服务器不知道什么时候开始向我们发送更新。一种方法是开始订阅，获得第一个更新，然后请求“更新 N 的状态”。这将需要服务器为每次更新存储一个快照，这是不切实际的。

图 59 - 状态复制

所以我们会在客户端做同步，如下：

客户端首先订阅更新，然后发出状态请求。这保证了状态将比它拥有的最旧的更新更新。
客户端等待服务器以状态回复，同时将所有更新排队。它只是通过不读取它们来做到这一点：ZeroMQ 使它们在套接字队列中排队。
当客户端收到它的状态更新时，它会再次开始读取更新。但是，它会丢弃任何比状态更新更旧的更新。因此，如果状态更新包括最多 200 次的更新，则客户端将丢弃最多 201 次的更新。
然后客户端将更新应用到它自己的状态快照。

这是一个利用 ZeroMQ 自己内部队列的简单模型。这是服务器：

//  Clone server - Model Two//  Lets us build this source without creating a library
#include "kvsimple.c"static int s_send_single (const char *key, void *data, void *args);
static void state_manager (void *args, zctx_t *ctx, void *pipe);int main (void)
{//  Prepare our context and socketszctx_t *ctx = zctx_new ();void *publisher = zsocket_new (ctx, ZMQ_PUB);zsocket_bind (publisher, "tcp://*:5557");int64_t sequence = 0;srandom ((unsigned) time (NULL));//  Start state manager and wait for synchronization signalvoid *updates = zthread_fork (ctx, state_manager, NULL);free (zstr_recv (updates));while (!zctx_interrupted) {//  Distribute as key-value messagekvmsg_t *kvmsg = kvmsg_new (++sequence);kvmsg_fmt_key  (kvmsg, "%d", randof (10000));kvmsg_fmt_body (kvmsg, "%d", randof (1000000));kvmsg_send     (kvmsg, publisher);kvmsg_send     (kvmsg, updates);kvmsg_destroy (&kvmsg);}printf (" Interrupted\n%d messages out\n", (int) sequence);zctx_destroy (&ctx);return 0;
}//  Routing information for a key-value snapshot
typedef struct {void *socket;           //  ROUTER socket to send tozframe_t *identity;     //  Identity of peer who requested state
} kvroute_t;//  Send one state snapshot key-value pair to a socket
//  Hash item data is our kvmsg object, ready to send
static int
s_send_single (const char *key, void *data, void *args)
{kvroute_t *kvroute = (kvroute_t *) args;//  Send identity of recipient firstzframe_send (&kvroute->identity,kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE);kvmsg_t *kvmsg = (kvmsg_t *) data;kvmsg_send (kvmsg, kvroute->socket);return 0;
}//  .split state manager
//  The state manager task maintains the state and handles requests from
//  clients for snapshots:static void
state_manager (void *args, zctx_t *ctx, void *pipe)
{zhash_t *kvmap = zhash_new ();zstr_send (pipe, "READY");void *snapshot = zsocket_new (ctx, ZMQ_ROUTER);zsocket_bind (snapshot, "tcp://*:5556");zmq_pollitem_t items [] = {{ pipe, 0, ZMQ_POLLIN, 0 },{ snapshot, 0, ZMQ_POLLIN, 0 }};int64_t sequence = 0;       //  Current snapshot version numberwhile (!zctx_interrupted) {int rc = zmq_poll (items, 2, -1);if (rc == -1 && errno == ETERM)break;              //  Context has been shut down//  Apply state update from main threadif (items [0].revents & ZMQ_POLLIN) {kvmsg_t *kvmsg = kvmsg_recv (pipe);if (!kvmsg)break;          //  Interruptedsequence = kvmsg_sequence (kvmsg);kvmsg_store (&kvmsg, kvmap);}//  Execute state snapshot requestif (items [1].revents & ZMQ_POLLIN) {zframe_t *identity = zframe_recv (snapshot);if (!identity)break;          //  Interrupted//  Request is in second frame of messagechar *request = zstr_recv (snapshot);if (streq (request, "ICANHAZ?"))free (request);else {printf ("E: bad request, aborting\n");break;}//  Send state snapshot to clientkvroute_t routing = { snapshot, identity };//  For each entry in kvmap, send kvmsg to clientzhash_foreach (kvmap, s_send_single, &routing);//  Now send END message with sequence numberprintf ("Sending state shapshot=%d\n", (int) sequence);zframe_send (&identity, snapshot, ZFRAME_MORE);kvmsg_t *kvmsg = kvmsg_new (sequence);kvmsg_set_key  (kvmsg, "KTHXBAI");kvmsg_set_body (kvmsg, (byte *) "", 0);kvmsg_send     (kvmsg, snapshot);kvmsg_destroy (&kvmsg);}}zhash_destroy (&kvmap);
}

这是客户端：

//  Clone client - Model Two//  Lets us build this source without creating a library
#include "kvsimple.c"int main (void)
{//  Prepare our context and subscriberzctx_t *ctx = zctx_new ();void *snapshot = zsocket_new (ctx, ZMQ_DEALER);zsocket_connect (snapshot, "tcp://localhost:5556");void *subscriber = zsocket_new (ctx, ZMQ_SUB);zsocket_set_subscribe (subscriber, "");zsocket_connect (subscriber, "tcp://localhost:5557");zhash_t *kvmap = zhash_new ();//  Get state snapshotint64_t sequence = 0;zstr_send (snapshot, "ICANHAZ?");while (true) {kvmsg_t *kvmsg = kvmsg_recv (snapshot);if (!kvmsg)break;          //  Interruptedif (streq (kvmsg_key (kvmsg), "KTHXBAI")) {sequence = kvmsg_sequence (kvmsg);printf ("Received snapshot=%d\n", (int) sequence);kvmsg_destroy (&kvmsg);break;          //  Done}kvmsg_store (&kvmsg, kvmap);}//  Now apply pending updates, discard out-of-sequence messageswhile (!zctx_interrupted) {kvmsg_t *kvmsg = kvmsg_recv (subscriber);if (!kvmsg)break;          //  Interruptedif (kvmsg_sequence (kvmsg) > sequence) {sequence = kvmsg_sequence (kvmsg);kvmsg_store (&kvmsg, kvmap);}elsekvmsg_destroy (&kvmsg);}zhash_destroy (&kvmap);zctx_destroy (&ctx);return 0;
}

以下是关于这两个程序的一些注意事项：

服务器使用两个任务。一个线程（随机）生成更新并将这些更新发送到主 PUB 套接字，而另一个线程处理 ROUTER 套接字上的状态请求。两者通过 inproc:@<//>@ 连接通过 PAIR 套接字进行通信。
客户端真的很简单。在 C 中，它由大约 50 行代码组成。许多繁重的工作都是在 kvmsg 类中完成的。即便如此，基本的克隆模式比最初看起来更容易实现。
我们不使用任何花哨的东西来序列化状态。哈希表包含一组 kvmsg 对象，服务器将这些对象作为一批消息发送到客户端请求状态。如果多个客户端同时请求状态，每个客户端将获得不同的快照。
我们假设客户端正好有一个服务器要与之通信。服务器必须正在运行；我们不会试图解决如果服务器崩溃会发生什么的问题

现在，这两个程序没有做任何实际的事情，但它们正确地同步了状态。这是如何混合不同模式的一个很好的例子：PAIR-PAIR、PUB-SUB 和 ROUTER-DEALER。

重新发布来自客户端的更新
在我们的第二个模型中，键值存储的更改来自服务器本身。这是一个很有用的集中式模型，例如，如果我们有一个要分发的中央配置文件，并且每个节点上都有本地缓存。一个更有趣的模型从客户端而不是服务器获取更新。服务器因此成为无状态代理。这给我们带来了一些好处：

我们不太担心服务器的可靠性。如果它崩溃了，我们可以启动一个新实例并为其提供新值。
我们可以使用键值存储在活跃对等点之间共享知识。

要将更新从客户端发送回服务器，我们可以使用各种套接字模式。最简单可行的解决方案是推拉组合。

为什么我们不允许客户端直接相互发布更新？虽然这会减少延迟，但会消除一致性的保证。如果允许更新的顺序根据接收者的不同而改变，则无法获得一致的共享状态。假设我们有两个客户端，正在更改不同的密钥。这将正常工作。但是如果两个客户端尝试大致同时更改同一个键，他们最终会得到不同的值概念。

当更改同时发生在多个地方时，有几种策略可以用来获得一致性。我们将使用集中所有更改的方法。无论客户端所做更改的准确时间如何，它们都会通过服务器推送，服务器根据获取更新的顺序强制执行单个序列。

图 60 - 重新发布更新

通过调解所有更改，服务器还可以为所有更新添加唯一的序列号。通过独特的排序，客户端可以检测到更严重的故障，包括网络拥塞和队列溢出。如果客户端发现它的传入消息流有一个漏洞，它可以采取行动。客户端联系服务器并询问丢失的消息似乎是明智的，但实际上这没有用。如果有漏洞，它们是由网络压力引起的，给网络增加更多的压力会使情况变得更糟。客户端所能做的就是警告它的用户它“无法继续”，停止，并且在有人手动检查问题原因之前不会重新启动。

我们现在将在客户端生成状态更新。这是服务器：

//  Clone server - Model Three//  Lets us build this source without creating a library
#include "kvsimple.c"//  Routing information for a key-value snapshot
typedef struct {void *socket;           //  ROUTER socket to send tozframe_t *identity;     //  Identity of peer who requested state
} kvroute_t;//  Send one state snapshot key-value pair to a socket
//  Hash item data is our kvmsg object, ready to send
static int
s_send_single (const char *key, void *data, void *args)
{kvroute_t *kvroute = (kvroute_t *) args;//  Send identity of recipient firstzframe_send (&kvroute->identity,kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE);kvmsg_t *kvmsg = (kvmsg_t *) data;kvmsg_send (kvmsg, kvroute->socket);return 0;
}int main (void)
{//  Prepare our context and socketszctx_t *ctx = zctx_new ();void *snapshot = zsocket_new (ctx, ZMQ_ROUTER);zsocket_bind (snapshot, "tcp://*:5556");void *publisher = zsocket_new (ctx, ZMQ_PUB);zsocket_bind (publisher, "tcp://*:5557");void *collector = zsocket_new (ctx, ZMQ_PULL);zsocket_bind (collector, "tcp://*:5558");//  .split body of main task//  The body of the main task collects updates from clients and//  publishes them back out to clients:int64_t sequence = 0;zhash_t *kvmap = zhash_new ();zmq_pollitem_t items [] = {{ collector, 0, ZMQ_POLLIN, 0 },{ snapshot, 0, ZMQ_POLLIN, 0 }};while (!zctx_interrupted) {int rc = zmq_poll (items, 2, 1000 * ZMQ_POLL_MSEC);//  Apply state update sent from clientif (items [0].revents & ZMQ_POLLIN) {kvmsg_t *kvmsg = kvmsg_recv (collector);if (!kvmsg)break;          //  Interruptedkvmsg_set_sequence (kvmsg, ++sequence);kvmsg_send (kvmsg, publisher);kvmsg_store (&kvmsg, kvmap);printf ("I: publishing update %5d\n", (int) sequence);}//  Execute state snapshot requestif (items [1].revents & ZMQ_POLLIN) {zframe_t *identity = zframe_recv (snapshot);if (!identity)break;          //  Interrupted//  Request is in second frame of messagechar *request = zstr_recv (snapshot);if (streq (request, "ICANHAZ?"))free (request);else {printf ("E: bad request, aborting\n");break;}//  Send state snapshot to clientkvroute_t routing = { snapshot, identity };//  For each entry in kvmap, send kvmsg to clientzhash_foreach (kvmap, s_send_single, &routing);//  Now send END message with sequence numberprintf ("I: sending shapshot=%d\n", (int) sequence);zframe_send (&identity, snapshot, ZFRAME_MORE);kvmsg_t *kvmsg = kvmsg_new (sequence);kvmsg_set_key  (kvmsg, "KTHXBAI");kvmsg_set_body (kvmsg, (byte *) "", 0);kvmsg_send     (kvmsg, snapshot);kvmsg_destroy (&kvmsg);}}printf (" Interrupted\n%d messages handled\n", (int) sequence);zhash_destroy (&kvmap);zctx_destroy (&ctx);return 0;
}

这是客户端：

//  Clone client - Model Three//  Lets us build this source without creating a library
#include "kvsimple.c"int main (void)
{//  Prepare our context and subscriberzctx_t *ctx = zctx_new ();void *snapshot = zsocket_new (ctx, ZMQ_DEALER);zsocket_connect (snapshot, "tcp://localhost:5556");void *subscriber = zsocket_new (ctx, ZMQ_SUB);zsocket_set_subscribe (subscriber, "");zsocket_connect (subscriber, "tcp://localhost:5557");void *publisher = zsocket_new (ctx, ZMQ_PUSH);zsocket_connect (publisher, "tcp://localhost:5558");zhash_t *kvmap = zhash_new ();srandom ((unsigned) time (NULL));//  .split getting a state snapshot//  We first request a state snapshot:int64_t sequence = 0;zstr_send (snapshot, "ICANHAZ?");while (true) {kvmsg_t *kvmsg = kvmsg_recv (snapshot);if (!kvmsg)break;          //  Interruptedif (streq (kvmsg_key (kvmsg), "KTHXBAI")) {sequence = kvmsg_sequence (kvmsg);printf ("I: received snapshot=%d\n", (int) sequence);kvmsg_destroy (&kvmsg);break;          //  Done}kvmsg_store (&kvmsg, kvmap);}//  .split processing state updates//  Now we wait for updates from the server and every so often, we//  send a random key-value update to the server:int64_t alarm = zclock_time () + 1000;while (!zctx_interrupted) {zmq_pollitem_t items [] = { { subscriber, 0, ZMQ_POLLIN, 0 } };int tickless = (int) ((alarm - zclock_time ()));if (tickless < 0)tickless = 0;int rc = zmq_poll (items, 1, tickless * ZMQ_POLL_MSEC);if (rc == -1)break;              //  Context has been shut downif (items [0].revents & ZMQ_POLLIN) {kvmsg_t *kvmsg = kvmsg_recv (subscriber);if (!kvmsg)break;          //  Interrupted//  Discard out-of-sequence kvmsgs, incl. heartbeatsif (kvmsg_sequence (kvmsg) > sequence) {sequence = kvmsg_sequence (kvmsg);kvmsg_store (&kvmsg, kvmap);printf ("I: received update=%d\n", (int) sequence);}elsekvmsg_destroy (&kvmsg);}//  If we timed out, generate a random kvmsgif (zclock_time () >= alarm) {kvmsg_t *kvmsg = kvmsg_new (0);kvmsg_fmt_key  (kvmsg, "%d", randof (10000));kvmsg_fmt_body (kvmsg, "%d", randof (1000000));kvmsg_send     (kvmsg, publisher);kvmsg_destroy (&kvmsg);alarm = zclock_time () + 1000;}}printf (" Interrupted\n%d messages in\n", (int) sequence);zhash_destroy (&kvmap);zctx_destroy (&ctx);return 0;
}

以下是有关第三种设计的一些注意事项：

服务器已折叠为单个任务。它管理用于传入更新的 PULL 套接字、用于状态请求的 ROUTER 套接字和用于传出更新的 PUB 套接字。
客户端使用一个简单的无滴答计时器每秒向服务器发送一次随机更新。在实际实现中，我们将从应用程序代码驱动更新。

使用子树 #
随着客户数量的增加，我们共享商店的规模也将扩大。将所有内容发送给每个客户不再合理。这是 pub-sub 的经典故事：当您的客户数量很少时，您可以将每条消息发送给所有客户。随着架构的增长，这变得低效。客户专注于不同的领域。

因此，即使在使用共享存储时，一些客户端也只希望使用该存储的一部分，我们称之为子树。客户端在发出状态请求时必须请求子树，并且在订阅更新时必须指定相同的子树。

树有几种常见的语法。一个是路径层次结构，另一个是主题树。这些看起来像这样：

路径层次结构：/some/list/of/paths
主题树：some.list.of.topics

我们将使用路径层次结构，并扩展我们的客户端和服务器，以便客户端可以使用单个子树。一旦您了解了如何使用单个子树，您就可以自己扩展它以处理多个子树，如果您的用例需要的话。

这是实现子树的服务器，模型三的一个小变化：

//  Clone server - Model Four//  Lets us build this source without creating a library
#include "kvsimple.c"//  Routing information for a key-value snapshot
typedef struct {void *socket;           //  ROUTER socket to send tozframe_t *identity;     //  Identity of peer who requested statechar *subtree;          //  Client subtree specification
} kvroute_t;//  Send one state snapshot key-value pair to a socket
//  Hash item data is our kvmsg object, ready to send
static int
s_send_single (const char *key, void *data, void *args)
{kvroute_t *kvroute = (kvroute_t *) args;kvmsg_t *kvmsg = (kvmsg_t *) data;if (strlen (kvroute->subtree) <= strlen (kvmsg_key (kvmsg))&&  memcmp (kvroute->subtree,kvmsg_key (kvmsg), strlen (kvroute->subtree)) == 0) {//  Send identity of recipient firstzframe_send (&kvroute->identity,kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE);kvmsg_send (kvmsg, kvroute->socket);}return 0;
}//  The main task is identical to clonesrv3 except for where it
//  handles subtrees.
//  .skipint main (void)
{//  Prepare our context and socketszctx_t *ctx = zctx_new ();void *snapshot = zsocket_new (ctx, ZMQ_ROUTER);zsocket_bind (snapshot, "tcp://*:5556");void *publisher = zsocket_new (ctx, ZMQ_PUB);zsocket_bind (publisher, "tcp://*:5557");void *collector = zsocket_new (ctx, ZMQ_PULL);zsocket_bind (collector, "tcp://*:5558");int64_t sequence = 0;zhash_t *kvmap = zhash_new ();zmq_pollitem_t items [] = {{ collector, 0, ZMQ_POLLIN, 0 },{ snapshot, 0, ZMQ_POLLIN, 0 }};while (!zctx_interrupted) {int rc = zmq_poll (items, 2, 1000 * ZMQ_POLL_MSEC);//  Apply state update sent from clientif (items [0].revents & ZMQ_POLLIN) {kvmsg_t *kvmsg = kvmsg_recv (collector);if (!kvmsg)break;          //  Interruptedkvmsg_set_sequence (kvmsg, ++sequence);kvmsg_send (kvmsg, publisher);kvmsg_store (&kvmsg, kvmap);printf ("I: publishing update %5d\n", (int) sequence);}//  Execute state snapshot requestif (items [1].revents & ZMQ_POLLIN) {zframe_t *identity = zframe_recv (snapshot);if (!identity)break;          //  Interrupted//  .until//  Request is in second frame of messagechar *request = zstr_recv (snapshot);char *subtree = NULL;if (streq (request, "ICANHAZ?")) {free (request);subtree = zstr_recv (snapshot);}//  .skipelse {printf ("E: bad request, aborting\n");break;}//  .until//  Send state snapshot to clientkvroute_t routing = { snapshot, identity, subtree };//  .skip//  For each entry in kvmap, send kvmsg to clientzhash_foreach (kvmap, s_send_single, &routing);//  .until//  Now send END message with sequence numberprintf ("I: sending shapshot=%d\n", (int) sequence);zframe_send (&identity, snapshot, ZFRAME_MORE);kvmsg_t *kvmsg = kvmsg_new (sequence);kvmsg_set_key  (kvmsg, "KTHXBAI");kvmsg_set_body (kvmsg, (byte *) subtree, 0);kvmsg_send     (kvmsg, snapshot);kvmsg_destroy (&kvmsg);free (subtree);}}//  .skipprintf (" Interrupted\n%d messages handled\n", (int) sequence);zhash_destroy (&kvmap);zctx_destroy (&ctx);return 0;
}

这是相应的客户端：

//  Clone client - Model Four//  Lets us build this source without creating a library
#include "kvsimple.c"//  This client is identical to clonecli3 except for where we
//  handles subtrees.
#define SUBTREE "/client/"
//  .skipint main (void)
{//  Prepare our context and subscriberzctx_t *ctx = zctx_new ();void *snapshot = zsocket_new (ctx, ZMQ_DEALER);zsocket_connect (snapshot, "tcp://localhost:5556");void *subscriber = zsocket_new (ctx, ZMQ_SUB);zsocket_set_subscribe (subscriber, "");//  .untilzsocket_connect (subscriber, "tcp://localhost:5557");zsocket_set_subscribe (subscriber, SUBTREE);//  .skipvoid *publisher = zsocket_new (ctx, ZMQ_PUSH);zsocket_connect (publisher, "tcp://localhost:5558");zhash_t *kvmap = zhash_new ();srandom ((unsigned) time (NULL));//  .until//  We first request a state snapshot:int64_t sequence = 0;zstr_sendm (snapshot, "ICANHAZ?");zstr_send  (snapshot, SUBTREE);//  .skipwhile (true) {kvmsg_t *kvmsg = kvmsg_recv (snapshot);if (!kvmsg)break;          //  Interruptedif (streq (kvmsg_key (kvmsg), "KTHXBAI")) {sequence = kvmsg_sequence (kvmsg);printf ("I: received snapshot=%d\n", (int) sequence);kvmsg_destroy (&kvmsg);break;          //  Done}kvmsg_store (&kvmsg, kvmap);}int64_t alarm = zclock_time () + 1000;while (!zctx_interrupted) {zmq_pollitem_t items [] = { { subscriber, 0, ZMQ_POLLIN, 0 } };int tickless = (int) ((alarm - zclock_time ()));if (tickless < 0)tickless = 0;int rc = zmq_poll (items, 1, tickless * ZMQ_POLL_MSEC);if (rc == -1)break;              //  Context has been shut downif (items [0].revents & ZMQ_POLLIN) {kvmsg_t *kvmsg = kvmsg_recv (subscriber);if (!kvmsg)break;          //  Interrupted//  Discard out-of-sequence kvmsgs, incl. heartbeatsif (kvmsg_sequence (kvmsg) > sequence) {sequence = kvmsg_sequence (kvmsg);kvmsg_store (&kvmsg, kvmap);printf ("I: received update=%d\n", (int) sequence);}elsekvmsg_destroy (&kvmsg);}//  .until//  If we timed out, generate a random kvmsgif (zclock_time () >= alarm) {kvmsg_t *kvmsg = kvmsg_new (0);kvmsg_fmt_key  (kvmsg, "%s%d", SUBTREE, randof (10000));kvmsg_fmt_body (kvmsg, "%d", randof (1000000));kvmsg_send     (kvmsg, publisher);kvmsg_destroy (&kvmsg);alarm = zclock_time () + 1000;}//  .skip}printf (" Interrupted\n%d messages in\n", (int) sequence);zhash_destroy (&kvmap);zctx_destroy (&ctx);return 0;
}

临时值
临时值是自动过期的值，除非定期刷新。如果您认为 Clone 用于注册服务，那么临时值可以让您执行动态值。一个节点加入网络，发布它的地址，并定期刷新它。如果节点死亡，它的地址最终会被删除。

通常抽象的将临时值附加到会话，并在会话结束时删除它们。在 Clone 中，会话将由客户端定义，如果客户端死亡，会话将结束。一个更简单的替代方法是将生存时间 (TTL) 附加到临时值，服务器使用它来使未及时刷新的值过期。

我尽可能使用的一个好的设计原则是不要发明不是绝对必要的概念。如果我们有大量的临时值，会话将提供更好的性能。如果我们使用少量的临时值，则可以为每个值设置一个 TTL。如果我们使用大量的临时值，将它们附加到会话并批量过期会更有效。这不是我们在现阶段面临的问题，也可能永远不会遇到，因此会话会消失。

现在我们将实现临时值。首先，我们需要一种方法来对键值消息中的 TTL 进行编码。我们可以添加一个框架。将 ZeroMQ 框架用于属性的问题在于，每次我们想要添加一个新属性时，我们都必须更改消息结构。它破坏了兼容性。所以让我们为消息添加一个属性框架，并编写代码让我们获取和放置属性值。

接下来，我们需要一种方式来表达，“删除这个值”。到目前为止，服务器和客户端总是盲目地将新值插入或更新到它们的哈希表中。我们会说，如果该值为空，则表示“删除此键”。

这是 kvmsg 类的更完整版本，它实现了属性框架（并添加了一个 UUID 框架，我们稍后会用到它）。如有必要，它还通过从哈希中删除键来处理空值：

//  kvmsg class - key-value message class for example applications#include "kvmsg.h"
#include <uuid/uuid.h>
#include "zlist.h"//  Keys are short strings
#define KVMSG_KEY_MAX   255//  Message is formatted on wire as 5 frames:
//  frame 0: key (0MQ string)
//  frame 1: sequence (8 bytes, network order)
//  frame 2: uuid (blob, 16 bytes)
//  frame 3: properties (0MQ string)
//  frame 4: body (blob)
#define FRAME_KEY       0
#define FRAME_SEQ       1
#define FRAME_UUID      2
#define FRAME_PROPS     3
#define FRAME_BODY      4
#define KVMSG_FRAMES    5//  Structure of our class
struct _kvmsg {//  Presence indicators for each frameint present [KVMSG_FRAMES];//  Corresponding 0MQ message frames, if anyzmq_msg_t frame [KVMSG_FRAMES];//  Key, copied into safe C stringchar key [KVMSG_KEY_MAX + 1];//  List of properties, as name=value stringszlist_t *props;size_t props_size;
};//  .split property encoding
//  These two helpers serialize a list of properties to and from a
//  message frame:static void
s_encode_props (kvmsg_t *self)
{zmq_msg_t *msg = &self->frame [FRAME_PROPS];if (self->present [FRAME_PROPS])zmq_msg_close (msg);zmq_msg_init_size (msg, self->props_size);char *prop = zlist_first (self->props);char *dest = (char *) zmq_msg_data (msg);while (prop) {strcpy (dest, prop);dest += strlen (prop);*dest++ = '\n';prop = zlist_next (self->props);}self->present [FRAME_PROPS] = 1;
}static void
s_decode_props (kvmsg_t *self)
{zmq_msg_t *msg = &self->frame [FRAME_PROPS];self->props_size = 0;while (zlist_size (self->props))free (zlist_pop (self->props));size_t remainder = zmq_msg_size (msg);char *prop = (char *) zmq_msg_data (msg);char *eoln = memchr (prop, '\n', remainder);while (eoln) {*eoln = 0;zlist_append (self->props, strdup (prop));self->props_size += strlen (prop) + 1;remainder -= strlen (prop) + 1;prop = eoln + 1;eoln = memchr (prop, '\n', remainder);}
}//  .split constructor and destructor
//  Here are the constructor and destructor for the class://  Constructor, takes a sequence number for the new kvmsg instance:
kvmsg_t *
kvmsg_new (int64_t sequence)
{kvmsg_t*self;self = (kvmsg_t *) zmalloc (sizeof (kvmsg_t));self->props = zlist_new ();kvmsg_set_sequence (self, sequence);return self;
}//  zhash_free_fn callback helper that does the low level destruction:
void
kvmsg_free (void *ptr)
{if (ptr) {kvmsg_t *self = (kvmsg_t *) ptr;//  Destroy message frames if anyint frame_nbr;for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++)if (self->present [frame_nbr])zmq_msg_close (&self->frame [frame_nbr]);//  Destroy property listwhile (zlist_size (self->props))free (zlist_pop (self->props));zlist_destroy (&self->props);//  Free object itselffree (self);}
}//  Destructor
void
kvmsg_destroy (kvmsg_t **self_p)
{assert (self_p);if (*self_p) {kvmsg_free (*self_p);*self_p = NULL;}
}//  .split recv method
//  This method reads a key-value message from the socket and returns a
//  new {{kvmsg}} instance:kvmsg_t *
kvmsg_recv (void *socket)
{//  This method is almost unchanged from kvsimple//  .skipassert (socket);kvmsg_t *self = kvmsg_new (0);//  Read all frames off the wire, reject if bogusint frame_nbr;for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++) {if (self->present [frame_nbr])zmq_msg_close (&self->frame [frame_nbr]);zmq_msg_init (&self->frame [frame_nbr]);self->present [frame_nbr] = 1;if (zmq_msg_recv (&self->frame [frame_nbr], socket, 0) == -1) {kvmsg_destroy (&self);break;}//  Verify multipart framingint rcvmore = (frame_nbr < KVMSG_FRAMES - 1)? 1: 0;if (zsocket_rcvmore (socket) != rcvmore) {kvmsg_destroy (&self);break;}}//  .untilif (self)s_decode_props (self);return self;
}//  Send key-value message to socket; any empty frames are sent as such.
void
kvmsg_send (kvmsg_t *self, void *socket)
{assert (self);assert (socket);s_encode_props (self);//  The rest of the method is unchanged from kvsimple//  .skipint frame_nbr;for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++) {zmq_msg_t copy;zmq_msg_init (&copy);if (self->present [frame_nbr])zmq_msg_copy (&copy, &self->frame [frame_nbr]);zmq_msg_send (&copy, socket,(frame_nbr < KVMSG_FRAMES - 1)? ZMQ_SNDMORE: 0);zmq_msg_close (&copy);}
}
//  .until//  .split dup method
//  This method duplicates a {{kvmsg}} instance, returns the new instance:kvmsg_t *
kvmsg_dup (kvmsg_t *self)
{kvmsg_t *kvmsg = kvmsg_new (0);int frame_nbr;for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++) {if (self->present [frame_nbr]) {zmq_msg_t *src = &self->frame [frame_nbr];zmq_msg_t *dst = &kvmsg->frame [frame_nbr];zmq_msg_init_size (dst, zmq_msg_size (src));memcpy (zmq_msg_data (dst),zmq_msg_data (src), zmq_msg_size (src));kvmsg->present [frame_nbr] = 1;}}kvmsg->props_size = zlist_size (self->props);char *prop = (char *) zlist_first (self->props);while (prop) {zlist_append (kvmsg->props, strdup (prop));prop = (char *) zlist_next (self->props);}return kvmsg;
}//  The key, sequence, body, and size methods are the same as in kvsimple.
//  .skip//  Return key from last read message, if any, else NULL
char *
kvmsg_key (kvmsg_t *self)
{assert (self);if (self->present [FRAME_KEY]) {if (!*self->key) {size_t size = zmq_msg_size (&self->frame [FRAME_KEY]);if (size > KVMSG_KEY_MAX)size = KVMSG_KEY_MAX;memcpy (self->key,zmq_msg_data (&self->frame [FRAME_KEY]), size);self->key [size] = 0;}return self->key;}elsereturn NULL;
}//  Set message key as provided
void
kvmsg_set_key (kvmsg_t *self, char *key)
{assert (self);zmq_msg_t *msg = &self->frame [FRAME_KEY];if (self->present [FRAME_KEY])zmq_msg_close (msg);zmq_msg_init_size (msg, strlen (key));memcpy (zmq_msg_data (msg), key, strlen (key));self->present [FRAME_KEY] = 1;
}//  Set message key using printf format
void
kvmsg_fmt_key (kvmsg_t *self, char *format, ...)
{char value [KVMSG_KEY_MAX + 1];va_list args;assert (self);va_start (args, format);vsnprintf (value, KVMSG_KEY_MAX, format, args);va_end (args);kvmsg_set_key (self, value);
}//  Return sequence nbr from last read message, if any
int64_t
kvmsg_sequence (kvmsg_t *self)
{assert (self);if (self->present [FRAME_SEQ]) {assert (zmq_msg_size (&self->frame [FRAME_SEQ]) == 8);byte *source = zmq_msg_data (&self->frame [FRAME_SEQ]);int64_t sequence = ((int64_t) (source [0]) << 56)+ ((int64_t) (source [1]) << 48)+ ((int64_t) (source [2]) << 40)+ ((int64_t) (source [3]) << 32)+ ((int64_t) (source [4]) << 24)+ ((int64_t) (source [5]) << 16)+ ((int64_t) (source [6]) << 8)+  (int64_t) (source [7]);return sequence;}elsereturn 0;
}//  Set message sequence number
void
kvmsg_set_sequence (kvmsg_t *self, int64_t sequence)
{assert (self);zmq_msg_t *msg = &self->frame [FRAME_SEQ];if (self->present [FRAME_SEQ])zmq_msg_close (msg);zmq_msg_init_size (msg, 8);byte *source = zmq_msg_data (msg);source [0] = (byte) ((sequence >> 56) & 255);source [1] = (byte) ((sequence >> 48) & 255);source [2] = (byte) ((sequence >> 40) & 255);source [3] = (byte) ((sequence >> 32) & 255);source [4] = (byte) ((sequence >> 24) & 255);source [5] = (byte) ((sequence >> 16) & 255);source [6] = (byte) ((sequence >> 8)  & 255);source [7] = (byte) ((sequence)       & 255);self->present [FRAME_SEQ] = 1;
}//  Return body from last read message, if any, else NULL
byte *
kvmsg_body (kvmsg_t *self)
{assert (self);if (self->present [FRAME_BODY])return (byte *) zmq_msg_data (&self->frame [FRAME_BODY]);elsereturn NULL;
}//  Set message body
void
kvmsg_set_body (kvmsg_t *self, byte *body, size_t size)
{assert (self);zmq_msg_t *msg = &self->frame [FRAME_BODY];if (self->present [FRAME_BODY])zmq_msg_close (msg);self->present [FRAME_BODY] = 1;zmq_msg_init_size (msg, size);memcpy (zmq_msg_data (msg), body, size);
}//  Set message body using printf format
void
kvmsg_fmt_body (kvmsg_t *self, char *format, ...)
{char value [255 + 1];va_list args;assert (self);va_start (args, format);vsnprintf (value, 255, format, args);va_end (args);kvmsg_set_body (self, (byte *) value, strlen (value));
}//  Return body size from last read message, if any, else zero
size_t
kvmsg_size (kvmsg_t *self)
{assert (self);if (self->present [FRAME_BODY])return zmq_msg_size (&self->frame [FRAME_BODY]);elsereturn 0;
}
//  .until//  .split UUID methods
//  These methods get and set the UUID for the key-value message:byte *
kvmsg_uuid (kvmsg_t *self)
{assert (self);if (self->present [FRAME_UUID]&&  zmq_msg_size (&self->frame [FRAME_UUID]) == sizeof (uuid_t))return (byte *) zmq_msg_data (&self->frame [FRAME_UUID]);elsereturn NULL;
}//  Sets the UUID to a randomly generated value
void
kvmsg_set_uuid (kvmsg_t *self)
{assert (self);zmq_msg_t *msg = &self->frame [FRAME_UUID];uuid_t uuid;uuid_generate (uuid);if (self->present [FRAME_UUID])zmq_msg_close (msg);zmq_msg_init_size (msg, sizeof (uuid));memcpy (zmq_msg_data (msg), uuid, sizeof (uuid));self->present [FRAME_UUID] = 1;
}//  .split property methods
//  These methods get and set a specified message property://  Get message property, return "" if no such property is defined.
char *
kvmsg_get_prop (kvmsg_t *self, char *name)
{assert (strchr (name, '=') == NULL);char *prop = zlist_first (self->props);size_t namelen = strlen (name);while (prop) {if (strlen (prop) > namelen&&  memcmp (prop, name, namelen) == 0&&  prop [namelen] == '=')return prop + namelen + 1;prop = zlist_next (self->props);}return "";
}//  Set message property. Property name cannot contain '='. Max length of
//  value is 255 chars.
void
kvmsg_set_prop (kvmsg_t *self, char *name, char *format, ...)
{assert (strchr (name, '=') == NULL);char value [255 + 1];va_list args;assert (self);va_start (args, format);vsnprintf (value, 255, format, args);va_end (args);//  Allocate name=value stringchar *prop = malloc (strlen (name) + strlen (value) + 2);//  Remove existing property if anysprintf (prop, "%s=", name);char *existing = zlist_first (self->props);while (existing) {if (memcmp (prop, existing, strlen (prop)) == 0) {self->props_size -= strlen (existing) + 1;zlist_remove (self->props, existing);free (existing);break;}existing = zlist_next (self->props);}//  Add new name=value property stringstrcat (prop, value);zlist_append (self->props, prop);self->props_size += strlen (prop) + 1;
}//  .split store method
//  This method stores the key-value message into a hash map, unless
//  the key and value are both null. It nullifies the {{kvmsg}} reference
//  so that the object is owned by the hash map, not the caller:void
kvmsg_store (kvmsg_t **self_p, zhash_t *hash)
{assert (self_p);if (*self_p) {kvmsg_t *self = *self_p;assert (self);if (kvmsg_size (self)) {if (self->present [FRAME_KEY]&&  self->present [FRAME_BODY]) {zhash_update (hash, kvmsg_key (self), self);zhash_freefn (hash, kvmsg_key (self), kvmsg_free);}}elsezhash_delete (hash, kvmsg_key (self));*self_p = NULL;}
}//  .split dump method
//  This method extends the {{kvsimple}} implementation with support for
//  message properties:void
kvmsg_dump (kvmsg_t *self)
{//  .skipif (self) {if (!self) {fprintf (stderr, "NULL");return;}size_t size = kvmsg_size (self);byte  *body = kvmsg_body (self);fprintf (stderr, "[seq:%" PRId64 "]", kvmsg_sequence (self));fprintf (stderr, "[key:%s]", kvmsg_key (self));//  .untilfprintf (stderr, "[size:%zd] ", size);if (zlist_size (self->props)) {fprintf (stderr, "[");char *prop = zlist_first (self->props);while (prop) {fprintf (stderr, "%s;", prop);prop = zlist_next (self->props);}fprintf (stderr, "]");}//  .skipint char_nbr;for (char_nbr = 0; char_nbr < size; char_nbr++)fprintf (stderr, "%02X", body [char_nbr]);fprintf (stderr, "\n");}elsefprintf (stderr, "NULL message\n");
}
//  .until//  .split test method
//  This method is the same as in {{kvsimple}} with added support
//  for the uuid and property features of {{kvmsg}}:int
kvmsg_test (int verbose)
{//  .skipkvmsg_t*kvmsg;printf (" * kvmsg: ");//  Prepare our context and socketszctx_t *ctx = zctx_new ();void *output = zsocket_new (ctx, ZMQ_DEALER);int rc = zmq_bind (output, "ipc://kvmsg_selftest.ipc");assert (rc == 0);void *input = zsocket_new (ctx, ZMQ_DEALER);rc = zmq_connect (input, "ipc://kvmsg_selftest.ipc");assert (rc == 0);zhash_t *kvmap = zhash_new ();//  .until//  Test send and receive of simple messagekvmsg = kvmsg_new (1);kvmsg_set_key  (kvmsg, "key");kvmsg_set_uuid (kvmsg);kvmsg_set_body (kvmsg, (byte *) "body", 4);if (verbose)kvmsg_dump (kvmsg);kvmsg_send (kvmsg, output);kvmsg_store (&kvmsg, kvmap);kvmsg = kvmsg_recv (input);if (verbose)kvmsg_dump (kvmsg);assert (streq (kvmsg_key (kvmsg), "key"));kvmsg_store (&kvmsg, kvmap);//  Test send and receive of message with propertieskvmsg = kvmsg_new (2);kvmsg_set_prop (kvmsg, "prop1", "value1");kvmsg_set_prop (kvmsg, "prop2", "value1");kvmsg_set_prop (kvmsg, "prop2", "value2");kvmsg_set_key  (kvmsg, "key");kvmsg_set_uuid (kvmsg);kvmsg_set_body (kvmsg, (byte *) "body", 4);assert (streq (kvmsg_get_prop (kvmsg, "prop2"), "value2"));if (verbose)kvmsg_dump (kvmsg);kvmsg_send (kvmsg, output);kvmsg_destroy (&kvmsg);kvmsg = kvmsg_recv (input);if (verbose)kvmsg_dump (kvmsg);assert (streq (kvmsg_key (kvmsg), "key"));assert (streq (kvmsg_get_prop (kvmsg, "prop2"), "value2"));kvmsg_destroy (&kvmsg);//  .skip//  Shutdown and destroy all objectszhash_destroy (&kvmap);zctx_destroy (&ctx);printf ("OK\n");return 0;
}
//  .until

模型五的客户几乎与模型四相同。它现在使用完整的 kvmsg 类，并在每条消息上设置一个随机的 ttl 属性（以秒为单位）：

kvmsg_set_prop (kvmsg, "ttl", "%d", randof (30));

使用反应器
到目前为止，我们在服务器中使用了轮询循环。在服务器的下一个模型中，我们切换到使用反应器。在 C 中，我们使用 CZMQ 的 zloop 类。使用反应器会使代码更冗长，但更容易理解和构建，因为服务器的每个部分都由单独的反应器处理程序处理。

我们使用单个线程并将服务器对象传递给反应器处理程序。我们可以将服务器组织成多个线程，每个线程处理一个套接字或定时器，但是当线程不必共享数据时效果更好。在这种情况下，所有工作都围绕服务器的哈希图进行，因此一个线程更简单。

有三个反应器处理程序：

一个处理来自 ROUTER 套接字的快照请求；
一个处理来自客户端的传入更新，来自 PULL 套接字；
一个使已通过其 TTL 的临时值过期。

clonesrv5: Clone server, Model Five in C
//  Clone server - Model Five//  Lets us build this source without creating a library
#include "kvmsg.c"//  zloop reactor handlers
static int s_snapshots (zloop_t *loop, zmq_pollitem_t *poller, void *args);
static int s_collector (zloop_t *loop, zmq_pollitem_t *poller, void *args);
static int s_flush_ttl (zloop_t *loop, int timer_id, void *args);//  Our server is defined by these properties
typedef struct {zctx_t *ctx;                //  Context wrapperzhash_t *kvmap;             //  Key-value storezloop_t *loop;              //  zloop reactorint port;                   //  Main port we're working onint64_t sequence;           //  How many updates we're atvoid *snapshot;             //  Handle snapshot requestsvoid *publisher;            //  Publish updates to clientsvoid *collector;            //  Collect updates from clients
} clonesrv_t;int main (void)
{clonesrv_t *self = (clonesrv_t *) zmalloc (sizeof (clonesrv_t));self->port = 5556;self->ctx = zctx_new ();self->kvmap = zhash_new ();self->loop = zloop_new ();zloop_set_verbose (self->loop, false);//  Set up our clone server socketsself->snapshot  = zsocket_new (self->ctx, ZMQ_ROUTER);zsocket_bind (self->snapshot,  "tcp://*:%d", self->port);self->publisher = zsocket_new (self->ctx, ZMQ_PUB);zsocket_bind (self->publisher, "tcp://*:%d", self->port + 1);self->collector = zsocket_new (self->ctx, ZMQ_PULL);zsocket_bind (self->collector, "tcp://*:%d", self->port + 2);//  Register our handlers with reactorzmq_pollitem_t poller = { 0, 0, ZMQ_POLLIN };poller.socket = self->snapshot;zloop_poller (self->loop, &poller, s_snapshots, self);poller.socket = self->collector;zloop_poller (self->loop, &poller, s_collector, self);zloop_timer (self->loop, 1000, 0, s_flush_ttl, self);//  Run reactor until process interruptedzloop_start (self->loop);zloop_destroy (&self->loop);zhash_destroy (&self->kvmap);zctx_destroy (&self->ctx);free (self);return 0;
}//  .split send snapshots
//  We handle ICANHAZ? requests by sending snapshot data to the
//  client that requested it://  Routing information for a key-value snapshot
typedef struct {void *socket;           //  ROUTER socket to send tozframe_t *identity;     //  Identity of peer who requested statechar *subtree;          //  Client subtree specification
} kvroute_t;//  We call this function for each key-value pair in our hash table
static int
s_send_single (const char *key, void *data, void *args)
{kvroute_t *kvroute = (kvroute_t *) args;kvmsg_t *kvmsg = (kvmsg_t *) data;if (strlen (kvroute->subtree) <= strlen (kvmsg_key (kvmsg))&&  memcmp (kvroute->subtree,kvmsg_key (kvmsg), strlen (kvroute->subtree)) == 0) {zframe_send (&kvroute->identity,    //  Choose recipientkvroute->socket, ZFRAME_MORE + ZFRAME_REUSE);kvmsg_send (kvmsg, kvroute->socket);}return 0;
}//  .split snapshot handler
//  This is the reactor handler for the snapshot socket; it accepts
//  just the ICANHAZ? request and replies with a state snapshot ending
//  with a KTHXBAI message:static int
s_snapshots (zloop_t *loop, zmq_pollitem_t *poller, void *args)
{clonesrv_t *self = (clonesrv_t *) args;zframe_t *identity = zframe_recv (poller->socket);if (identity) {//  Request is in second frame of messagechar *request = zstr_recv (poller->socket);char *subtree = NULL;if (streq (request, "ICANHAZ?")) {free (request);subtree = zstr_recv (poller->socket);}elseprintf ("E: bad request, aborting\n");if (subtree) {//  Send state socket to clientkvroute_t routing = { poller->socket, identity, subtree };zhash_foreach (self->kvmap, s_send_single, &routing);//  Now send END message with sequence numberzclock_log ("I: sending shapshot=%d", (int) self->sequence);zframe_send (&identity, poller->socket, ZFRAME_MORE);kvmsg_t *kvmsg = kvmsg_new (self->sequence);kvmsg_set_key  (kvmsg, "KTHXBAI");kvmsg_set_body (kvmsg, (byte *) subtree, 0);kvmsg_send     (kvmsg, poller->socket);kvmsg_destroy (&kvmsg);free (subtree);}zframe_destroy(&identity);}return 0;
}//  .split collect updates
//  We store each update with a new sequence number, and if necessary, a
//  time-to-live. We publish updates immediately on our publisher socket:static int
s_collector (zloop_t *loop, zmq_pollitem_t *poller, void *args)
{clonesrv_t *self = (clonesrv_t *) args;kvmsg_t *kvmsg = kvmsg_recv (poller->socket);if (kvmsg) {kvmsg_set_sequence (kvmsg, ++self->sequence);kvmsg_send (kvmsg, self->publisher);int ttl = atoi (kvmsg_get_prop (kvmsg, "ttl"));if (ttl)kvmsg_set_prop (kvmsg, "ttl","%" PRId64, zclock_time () + ttl * 1000);kvmsg_store (&kvmsg, self->kvmap);zclock_log ("I: publishing update=%d", (int) self->sequence);}return 0;
}//  .split flush ephemeral values
//  At regular intervals, we flush ephemeral values that have expired. This
//  could be slow on very large data sets://  If key-value pair has expired, delete it and publish the
//  fact to listening clients.
static int
s_flush_single (const char *key, void *data, void *args)
{clonesrv_t *self = (clonesrv_t *) args;kvmsg_t *kvmsg = (kvmsg_t *) data;int64_t ttl;sscanf (kvmsg_get_prop (kvmsg, "ttl"), "%" PRId64, &ttl);if (ttl && zclock_time () >= ttl) {kvmsg_set_sequence (kvmsg, ++self->sequence);kvmsg_set_body (kvmsg, (byte *) "", 0);kvmsg_send (kvmsg, self->publisher);kvmsg_store (&kvmsg, self->kvmap);zclock_log ("I: publishing delete=%d", (int) self->sequence);}return 0;
}static int
s_flush_ttl (zloop_t *loop, int timer_id, void *args)
{clonesrv_t *self = (clonesrv_t *) args;if (self->kvmap)zhash_foreach (self->kvmap, s_flush_single, args);return 0;
}

到目前为止，我们探索的克隆模型相对简单。现在我们将进入令人不快的复杂领域，这让我起床再喝一杯浓缩咖啡。您应该意识到，制作“可靠”的消息传递非常复杂，以至于您总是需要问：“我们真的需要这个吗？”在跳进去之前。如果您能够摆脱不可靠或“足够好”的可靠性，您就可以在成本和复杂性方面取得巨大成功。当然，您有时可能会丢失一些数据。这通常是一个很好的权衡。话虽如此，还有……啜饮……因为浓缩咖啡真的很好，让我们开始吧。

当您使用最后一个模型时，您将停止并重新启动服务器。它可能看起来像是恢复了，但当然它正在将更新应用于空状态而不是正确的当前状态。任何加入网络的新客户端只会获得最新的更新，而不是完整的历史记录。

我们想要的是一种让服务器从被杀死或崩溃中恢复的方法。我们还需要提供备份，以防服务器在任何时间段内无法使用。当有人要求“可靠性”时，让他们列出他们想要处理的故障。在我们的例子中，这些是：

服务器进程崩溃并自动或手动重新启动。这个过程失去了它的状态，必须从某个地方把它找回来。
服务器机器死机并长时间处于脱机状态。客户端必须切换到某个地方的备用服务器。
服务器进程或机器与网络断开连接，例如，交换机死机或数据中心被淘汰。它可能会在某个时候回来，但同时客户端需要一个备用服务器。

我们的第一步是添加第二台服务器。我们可以使用第 4 章 - 可靠的请求-回复模式中的二进制星形模式将它们组织成主要和备份。 Binary Star 是一个反应器，所以我们已经将最后一个服务器模型重构为一个反应器风格是很有用的。

如果主服务器崩溃，我们需要确保更新不会丢失。最简单的技术是将它们发送到两个服务器。然后，备份服务器可以充当客户端，并像所有客户端一样通过接收更新来保持其状态同步。它还将从客户那里获得新的更新。它还不能将这些存储在它的哈希表中，但它可以保留它们一段时间。

因此，模型六相对于模型五引入了以下变化：

对于发送到服务器的客户端更新，我们使用发布-订阅流而不是推拉流。这负责将更新分发到两台服务器。否则我们将不得不使用两个 DEALER 套接字。
我们将心跳添加到服务器更新（到客户端），以便客户端可以检测到主服务器何时死亡。然后它可以切换到备份服务器。
我们使用 Binary Star bstar reactor 类连接两个服务器。 Binary Star 依赖客户端通过向他们认为处于活动状态的服务器发出明确请求来进行投票。我们将使用快照请求作为投票机制。
我们通过添加 UUID 字段使所有更新消息都可唯一识别。客户端生成它，服务器将它传播回重新发布的更新。
被动服务器保留一个“待处理列表”，其中包含它已从客户端收到但尚未从主动服务器收到的更新；或更新它是从活动服务器收到的，但尚未从客户端收到。该列表按从最旧到最新的顺序排列，因此很容易从头上删除更新。

图 61 - 克隆客户端有限状态机

将客户端逻辑设计为有限状态机很有用。客户端循环通过三种状态：

客户端打开并连接其套接字，然后从第一个服务器请求快照。为了避免请求风暴，它只会询问任何给定的服务器两次。一个请求可能会丢失，这将是厄运。两个丢失就是粗心大意。
客户端等待来自当前服务器的回复（快照数据），如果收到，则将其存储。如果在一段时间内没有回复，它就会故障转移到下一个服务器。
当客户端获得它的快照时，它等待并处理更新。同样，如果它在一段时间内没有收到来自服务器的任何消息，它就会故障转移到下一个服务器。

客户端永远循环。在启动或故障转移期间，很可能一些客户端尝试与主服务器通信，而其他客户端尝试与备份服务器通信。 Binary Star 状态机处理这个，希望是准确的。很难证明软件是正确的；相反，我们会敲打它，直到我们无法证明它是错误的。

故障转移发生如下：

客户端检测到主服务器不再发送心跳，并断定它已经死了。客户端连接到备份服务器并请求一个新的状态快照。
备份服务器开始接收来自客户端的快照请求，并检测到主服务器已经离开，因此它作为主服务器接管。
备份服务器将其挂起列表应用于自己的哈希表，然后开始处理状态快照请求。

当主服务器重新联机时，它将：

作为被动服务器启动，并作为克隆客户端连接到备份服务器。
开始通过其 SUB 套接字从客户端接收更新。

我们做几个假设：

至少一台服务器将继续运行。如果两台服务器都崩溃，我们将丢失所有服务器状态，并且无法恢复。
多个客户端不会同时更新相同的哈希键。客户端更新将以不同的顺序到达两个服务器。因此，备份服务器可能会以不同于主服务器将要或所做的顺序应用其挂起列表中的更新。来自一个客户端的更新将始终以相同的顺序到达两台服务器，因此是安全的。

因此，我们使用二进制星型模式的高可用性服务器对的架构有两个服务器和一组与这两个服务器通信的客户端。

图 62 - 高可用性克隆服务器对

这是克隆服务器的第六个也是最后一个模型：

//  Clone server Model Six//  Lets us build this source without creating a library
#include "bstar.c"
#include "kvmsg.c"//  .split definitions
//  We define a set of reactor handlers and our server object structure://  Bstar reactor handlers
static ints_snapshots   (zloop_t *loop, zmq_pollitem_t *poller, void *args);
static ints_collector   (zloop_t *loop, zmq_pollitem_t *poller, void *args);
static ints_flush_ttl   (zloop_t *loop, int timer_id, void *args);
static ints_send_hugz   (zloop_t *loop, int timer_id, void *args);
static ints_new_active  (zloop_t *loop, zmq_pollitem_t *poller, void *args);
static ints_new_passive (zloop_t *loop, zmq_pollitem_t *poller, void *args);
static ints_subscriber  (zloop_t *loop, zmq_pollitem_t *poller, void *args);//  Our server is defined by these properties
typedef struct {zctx_t *ctx;                //  Context wrapperzhash_t *kvmap;             //  Key-value storebstar_t *bstar;             //  Bstar reactor coreint64_t sequence;           //  How many updates we're atint port;                   //  Main port we're working onint peer;                   //  Main port of our peervoid *publisher;            //  Publish updates and hugzvoid *collector;            //  Collect updates from clientsvoid *subscriber;           //  Get updates from peerzlist_t *pending;           //  Pending updates from clientsbool primary;               //  true if we're primarybool active;                //  true if we're activebool passive;               //  true if we're passive
} clonesrv_t;//  .split main task setup
//  The main task parses the command line to decide whether to start
//  as a primary or backup server. We're using the Binary Star pattern
//  for reliability. This interconnects the two servers so they can
//  agree on which one is primary and which one is backup. To allow the
//  two servers to run on the same box, we use different ports for
//  primary and backup. Ports 5003/5004 are used to interconnect the
//  servers. Ports 5556/5566 are used to receive voting events (snapshot
//  requests in the clone pattern). Ports 5557/5567 are used by the
//  publisher, and ports 5558/5568 are used by the collector:int main (int argc, char *argv [])
{clonesrv_t *self = (clonesrv_t *) zmalloc (sizeof (clonesrv_t));if (argc == 2 && streq (argv [1], "-p")) {zclock_log ("I: primary active, waiting for backup (passive)");self->bstar = bstar_new (BSTAR_PRIMARY, "tcp://*:5003","tcp://localhost:5004");bstar_voter (self->bstar, "tcp://*:5556",ZMQ_ROUTER, s_snapshots, self);self->port = 5556;self->peer = 5566;self->primary = true;}elseif (argc == 2 && streq (argv [1], "-b")) {zclock_log ("I: backup passive, waiting for primary (active)");self->bstar = bstar_new (BSTAR_BACKUP, "tcp://*:5004","tcp://localhost:5003");bstar_voter (self->bstar, "tcp://*:5566",ZMQ_ROUTER, s_snapshots, self);self->port = 5566;self->peer = 5556;self->primary = false;}else {printf ("Usage: clonesrv6 { -p | -b }\n");free (self);exit (0);}//  Primary server will become first activeif (self->primary)self->kvmap = zhash_new ();self->ctx = zctx_new ();self->pending = zlist_new ();bstar_set_verbose (self->bstar, true);//  Set up our clone server socketsself->publisher = zsocket_new (self->ctx, ZMQ_PUB);self->collector = zsocket_new (self->ctx, ZMQ_SUB);zsocket_set_subscribe (self->collector, "");zsocket_bind (self->publisher, "tcp://*:%d", self->port + 1);zsocket_bind (self->collector, "tcp://*:%d", self->port + 2);//  Set up our own clone client interface to peerself->subscriber = zsocket_new (self->ctx, ZMQ_SUB);zsocket_set_subscribe (self->subscriber, "");zsocket_connect (self->subscriber,"tcp://localhost:%d", self->peer + 1);//  .split main task body//  After we've setup our sockets, we register our binary star//  event handlers, and then start the bstar reactor. This finishes//  when the user presses Ctrl-C or when the process receives a SIGINT//  interrupt://  Register state change handlersbstar_new_active (self->bstar, s_new_active, self);bstar_new_passive (self->bstar, s_new_passive, self);//  Register our other handlers with the bstar reactorzmq_pollitem_t poller = { self->collector, 0, ZMQ_POLLIN };zloop_poller (bstar_zloop (self->bstar), &poller, s_collector, self);zloop_timer  (bstar_zloop (self->bstar), 1000, 0, s_flush_ttl, self);zloop_timer  (bstar_zloop (self->bstar), 1000, 0, s_send_hugz, self);//  Start the bstar reactorbstar_start (self->bstar);//  Interrupted, so shut downwhile (zlist_size (self->pending)) {kvmsg_t *kvmsg = (kvmsg_t *) zlist_pop (self->pending);kvmsg_destroy (&kvmsg);}zlist_destroy (&self->pending);bstar_destroy (&self->bstar);zhash_destroy (&self->kvmap);zctx_destroy (&self->ctx);free (self);return 0;
}//  We handle ICANHAZ? requests exactly as in the clonesrv5 example.
//  .skip//  Routing information for a key-value snapshot
typedef struct {void *socket;           //  ROUTER socket to send tozframe_t *identity;     //  Identity of peer who requested statechar *subtree;          //  Client subtree specification
} kvroute_t;//  Send one state snapshot key-value pair to a socket
//  Hash item data is our kvmsg object, ready to send
static int
s_send_single (const char *key, void *data, void *args)
{kvroute_t *kvroute = (kvroute_t *) args;kvmsg_t *kvmsg = (kvmsg_t *) data;if (strlen (kvroute->subtree) <= strlen (kvmsg_key (kvmsg))&&  memcmp (kvroute->subtree,kvmsg_key (kvmsg), strlen (kvroute->subtree)) == 0) {zframe_send (&kvroute->identity,    //  Choose recipientkvroute->socket, ZFRAME_MORE + ZFRAME_REUSE);kvmsg_send (kvmsg, kvroute->socket);}return 0;
}static int
s_snapshots (zloop_t *loop, zmq_pollitem_t *poller, void *args)
{clonesrv_t *self = (clonesrv_t *) args;zframe_t *identity = zframe_recv (poller->socket);if (identity) {//  Request is in second frame of messagechar *request = zstr_recv (poller->socket);char *subtree = NULL;if (streq (request, "ICANHAZ?")) {free (request);subtree = zstr_recv (poller->socket);}elseprintf ("E: bad request, aborting\n");if (subtree) {//  Send state socket to clientkvroute_t routing = { poller->socket, identity, subtree };zhash_foreach (self->kvmap, s_send_single, &routing);//  Now send END message with sequence numberzclock_log ("I: sending shapshot=%d", (int) self->sequence);zframe_send (&identity, poller->socket, ZFRAME_MORE);kvmsg_t *kvmsg = kvmsg_new (self->sequence);kvmsg_set_key  (kvmsg, "KTHXBAI");kvmsg_set_body (kvmsg, (byte *) subtree, 0);kvmsg_send     (kvmsg, poller->socket);kvmsg_destroy (&kvmsg);free (subtree);}zframe_destroy(&identity);}return 0;
}
//  .until//  .split collect updates
//  The collector is more complex than in the clonesrv5 example because the
//  way it processes updates depends on whether we're active or passive.
//  The active applies them immediately to its kvmap, whereas the passive
//  queues them as pending://  If message was already on pending list, remove it and return true,
//  else return false.
static int
s_was_pending (clonesrv_t *self, kvmsg_t *kvmsg)
{kvmsg_t *held = (kvmsg_t *) zlist_first (self->pending);while (held) {if (memcmp (kvmsg_uuid (kvmsg),kvmsg_uuid (held), sizeof (uuid_t)) == 0) {zlist_remove (self->pending, held);return true;}held = (kvmsg_t *) zlist_next (self->pending);}return false;
}static int
s_collector (zloop_t *loop, zmq_pollitem_t *poller, void *args)
{clonesrv_t *self = (clonesrv_t *) args;kvmsg_t *kvmsg = kvmsg_recv (poller->socket);if (kvmsg) {if (self->active) {kvmsg_set_sequence (kvmsg, ++self->sequence);kvmsg_send (kvmsg, self->publisher);int ttl = atoi (kvmsg_get_prop (kvmsg, "ttl"));if (ttl)kvmsg_set_prop (kvmsg, "ttl","%" PRId64, zclock_time () + ttl * 1000);kvmsg_store (&kvmsg, self->kvmap);zclock_log ("I: publishing update=%d", (int) self->sequence);}else {//  If we already got message from active, drop it, else//  hold on pending listif (s_was_pending (self, kvmsg))kvmsg_destroy (&kvmsg);elsezlist_append (self->pending, kvmsg);}}return 0;
}//  We purge ephemeral values using exactly the same code as in
//  the previous clonesrv5 example.
//  .skip
//  If key-value pair has expired, delete it and publish the
//  fact to listening clients.
static int
s_flush_single (const char *key, void *data, void *args)
{clonesrv_t *self = (clonesrv_t *) args;kvmsg_t *kvmsg = (kvmsg_t *) data;int64_t ttl;sscanf (kvmsg_get_prop (kvmsg, "ttl"), "%" PRId64, &ttl);if (ttl && zclock_time () >= ttl) {kvmsg_set_sequence (kvmsg, ++self->sequence);kvmsg_set_body (kvmsg, (byte *) "", 0);kvmsg_send (kvmsg, self->publisher);kvmsg_store (&kvmsg, self->kvmap);zclock_log ("I: publishing delete=%d", (int) self->sequence);}return 0;
}static int
s_flush_ttl (zloop_t *loop, int timer_id, void *args)
{clonesrv_t *self = (clonesrv_t *) args;if (self->kvmap)zhash_foreach (self->kvmap, s_flush_single, args);return 0;
}
//  .until//  .split heartbeating
//  We send a HUGZ message once a second to all subscribers so that they
//  can detect if our server dies. They'll then switch over to the backup
//  server, which will become active:static int
s_send_hugz (zloop_t *loop, int timer_id, void *args)
{clonesrv_t *self = (clonesrv_t *) args;kvmsg_t *kvmsg = kvmsg_new (self->sequence);kvmsg_set_key  (kvmsg, "HUGZ");kvmsg_set_body (kvmsg, (byte *) "", 0);kvmsg_send     (kvmsg, self->publisher);kvmsg_destroy (&kvmsg);return 0;
}//  .split handling state changes
//  When we switch from passive to active, we apply our pending list so that
//  our kvmap is up-to-date. When we switch to passive, we wipe our kvmap
//  and grab a new snapshot from the active server:static int
s_new_active (zloop_t *loop, zmq_pollitem_t *unused, void *args)
{clonesrv_t *self = (clonesrv_t *) args;self->active = true;self->passive = false;//  Stop subscribing to updateszmq_pollitem_t poller = { self->subscriber, 0, ZMQ_POLLIN };zloop_poller_end (bstar_zloop (self->bstar), &poller);//  Apply pending list to own hash tablewhile (zlist_size (self->pending)) {kvmsg_t *kvmsg = (kvmsg_t *) zlist_pop (self->pending);kvmsg_set_sequence (kvmsg, ++self->sequence);kvmsg_send (kvmsg, self->publisher);kvmsg_store (&kvmsg, self->kvmap);zclock_log ("I: publishing pending=%d", (int) self->sequence);}return 0;
}static int
s_new_passive (zloop_t *loop, zmq_pollitem_t *unused, void *args)
{clonesrv_t *self = (clonesrv_t *) args;zhash_destroy (&self->kvmap);self->active = false;self->passive = true;//  Start subscribing to updateszmq_pollitem_t poller = { self->subscriber, 0, ZMQ_POLLIN };zloop_poller (bstar_zloop (self->bstar), &poller, s_subscriber, self);return 0;
}//  .split subscriber handler
//  When we get an update, we create a new kvmap if necessary, and then
//  add our update to our kvmap. We're always passive in this case:static int
s_subscriber (zloop_t *loop, zmq_pollitem_t *poller, void *args)
{clonesrv_t *self = (clonesrv_t *) args;//  Get state snapshot if necessaryif (self->kvmap == NULL) {self->kvmap = zhash_new ();void *snapshot = zsocket_new (self->ctx, ZMQ_DEALER);zsocket_connect (snapshot, "tcp://localhost:%d", self->peer);zclock_log ("I: asking for snapshot from: tcp://localhost:%d",self->peer);zstr_sendm (snapshot, "ICANHAZ?");zstr_send (snapshot, ""); // blank subtree to get allwhile (true) {kvmsg_t *kvmsg = kvmsg_recv (snapshot);if (!kvmsg)break;          //  Interruptedif (streq (kvmsg_key (kvmsg), "KTHXBAI")) {self->sequence = kvmsg_sequence (kvmsg);kvmsg_destroy (&kvmsg);break;          //  Done}kvmsg_store (&kvmsg, self->kvmap);}zclock_log ("I: received snapshot=%d", (int) self->sequence);zsocket_destroy (self->ctx, snapshot);}//  Find and remove update off pending listkvmsg_t *kvmsg = kvmsg_recv (poller->socket);if (!kvmsg)return 0;if (strneq (kvmsg_key (kvmsg), "HUGZ")) {if (!s_was_pending (self, kvmsg)) {//  If active update came before client update, flip it//  around, store active update (with sequence) on pending//  list and use to clear client update when it comes laterzlist_append (self->pending, kvmsg_dup (kvmsg));}//  If update is more recent than our kvmap, apply itif (kvmsg_sequence (kvmsg) > self->sequence) {self->sequence = kvmsg_sequence (kvmsg);kvmsg_store (&kvmsg, self->kvmap);zclock_log ("I: received update=%d", (int) self->sequence);}elsekvmsg_destroy (&kvmsg);}elsekvmsg_destroy (&kvmsg);return 0;
}

这个模型只有几百行代码，但花了很长时间才开始工作。准确地说，构建 Model 6 花了大约整整一周的时间进行“天哪，这对于一个例子来说太复杂了”黑客攻击。我们已经组装了几乎所有东西，厨房也融入了这个小应用程序。我们有故障转移、临时值、子树等等。令我惊讶的是，前期设计非常准确。编写和调试这么多套接字流的细节仍然非常具有挑战性。

基于反应器的设计从代码中移除了很多繁重的工作，剩下的更简单、更容易理解。我们重用了第 4 章 - 可靠的请求-回复模式中的 bstar 反应器。整个服务器作为一个线程运行，所以没有线程间奇怪的事情发生——只是一个结构指针（self）传递给所有处理程序，它们可以愉快地做他们的事情。使用反应器的一个很好的副作用是代码不太紧密地集成到轮询循环中，更容易重用。模型六的大块取自模型五。

我一块一块地建造它，并在进入下一块之前让每一块都正常工作。因为有四五个主要的套接字流，这意味着需要大量的调试和测试。我只是通过将消息转储到控制台进行调试。不要使用经典的调试器来单步调试 ZeroMQ 应用程序；您需要查看消息流以了解正在发生的事情。

对于测试，我总是尝试使用 Valgrind，它可以捕获内存泄漏和无效内存访问。在 C 中，这是一个主要问题，因为您不能委托给垃圾收集器。使用像 kvmsg 和 CZMQ 这样的正确和一致的抽象有很大帮助。

集群哈希映射协议
虽然服务器几乎是先前模型加上二进制星形模式的混搭，但客户端要复杂得多。但在我们开始之前，让我们看看最终的协议。我在 ZeroMQ RFC 网站上将其写成集群哈希映射协议的规范。

粗略地说，有两种方法可以设计像这样的复杂协议。一种方法是将每个流分成自己的一组套接字。这是我们在这里使用的方法。优点是每个流程都简单干净。缺点是一次管理多个套接字流可能非常复杂。使用反应器使它变得更简单，但它仍然会产生许多必须正确组装在一起的移动部件。

第二种创建这种协议的方法是对所有内容使用单个套接字对。在这种情况下，我将 ROUTER 用于服务器，DEALER 用于客户端，然后通过该连接完成所有工作。它产生了一个更复杂的协议，但至少复杂性都集中在一个地方。在第 7 章 - 使用 ZeroMQ 的高级架构中，我们将看一个通过 ROUTER-DEALER 组合完成的协议示例。

让我们来看看 CHP 规范。请注意，“应该”、“必须”和“可以”是我们在协议规范中用来表示需求级别的关键词。

目标

CHP 旨在为通过 ZeroMQ 网络连接的客户端集群提供可靠的发布-订阅基础。它定义了一个由键值对组成的“hashmap”抽象。任何客户端都可以随时修改任何键值对，并且更改会传播到所有客户端。客户端可以随时加入网络。

建筑学

CHP 连接一组客户端应用程序和一组服务器。客户端连接到服务器。客户看不到对方。客户可以随意进出。

端口和连接

服务器必须打开三个端口，如下所示：

端口号 P 处的 SNAPSHOT 端口（ZeroMQ ROUTER 套接字）。
端口号为 P + 1 的 PUBLISHER 端口（ZeroMQ PUB 套接字）。
端口号 P + 2 处的 COLLECTOR 端口（ZeroMQ SUB 套接字）。

客户端应该至少打开两个连接：

到端口号 P 的快照连接（ZeroMQ DEALER 套接字）。
到端口号 P + 1 的订阅者连接（ZeroMQ SUB 套接字）。

客户端可以打开第三个连接，如果它想更新哈希图：

到端口号 P + 2 的 PUBLISHER 连接（ZeroMQ PUB 套接字）。

这个额外的帧没有显示在下面解释的命令中。

状态同步

客户端必须首先向其快照连接发送 ICANHAZ 命令。该命令由两帧组成，如下所示：

ICAHAZ 命令
---------------------
第 0 帧：“ICANHAZ？”
第一帧：子树规范

两个帧都是 ZeroMQ 字符串。子树规范可能为空。如果不为空，则它由一个斜杠后跟一个或多个路径段组成，以斜杠结尾。

服务器必须通过向其快照端口发送零个或多个 KVSYNC 命令来响应 ICANHAZ 命令，然后是 KTHXBAI 命令。服务器必须在每个命令前面加上客户端的身份，正如 ZeroMQ 和 ICANHAZ 命令所提供的那样。 KVSYNC 命令指定单个键值对，如下所示：

KVSYNC command
-----------------------------------
Frame 0: key, as ZeroMQ string
Frame 1: sequence number, 8 bytes in network order
Frame 2: <empty>
Frame 3: <empty>
Frame 4: value, as blob

序列号没有意义，可能为零。

KTHXBAI 命令采用以下形式：

KTHXBAI command
-----------------------------------
Frame 0: "KTHXBAI"
Frame 1: sequence number, 8 bytes in network order
Frame 2: <empty>
Frame 3: <empty>
Frame 4: subtree specification

序列号必须是之前发送的 KVSYNC 命令的最高序列号。

当客户端收到 KTHXBAI 命令时，它应该开始从其订阅者连接接收消息并应用它们。

服务器到客户端更新

当服务器对其哈希图进行更新时，它必须将其作为 KVPUB 命令在其发布者套接字上广播。 KVPUB 命令具有以下形式：

KVPUB command
-----------------------------------
Frame 0: key, as ZeroMQ string
Frame 1: sequence number, 8 bytes in network order
Frame 2: UUID, 16 bytes
Frame 3: properties, as ZeroMQ string
Frame 4: value, as blob

序列号必须严格递增。客户端必须丢弃任何序列号不严格大于最后接收到的 KTHXBAI 或 KVPUB 命令的 KVPUB 命令。

UUID 是可选的，第 2 帧可以为空（大小为零）。属性字段的格式为零个或多个“name=value”实例，后跟换行符。如果键值对没有属性，则属性字段为空。

如果该值为空，则客户端应该删除其具有指定键的键值条目。

在没有其他更新的情况下，服务器应该定期发送一个 HUGZ 命令，例如，每秒一次。 HUGZ 命令具有以下格式：

HUGZ command
-----------------------------------
Frame 0: "HUGZ"
Frame 1: 00000000
Frame 2: <empty>
Frame 3: <empty>
Frame 4: <empty>

客户端可以将 HUGZ 的缺失视为服务器崩溃的指示器（参见下面的可靠性）。

客户端到服务器更新

当客户端对其哈希映射进行更新时，它可以将其作为 KVSET 命令通过其发布者连接发送到服务器。 KVSET 命令具有以下形式：

KVSET command
-----------------------------------
Frame 0: key, as ZeroMQ string
Frame 1: sequence number, 8 bytes in network order
Frame 2: UUID, 16 bytes
Frame 3: properties, as ZeroMQ string
Frame 4: value, as blob

序列号没有意义，可能为零。如果使用可靠的服务器架构，UUID 应该是通用唯一标识符。

如果该值为空，则服务器必须删除其具有指定键的键值条目。

服务器应该接受以下属性：

ttl：以秒为单位指定生存时间。如果 KVSET 命令有一个 ttl 属性，服务器应该删除键值对并广播一个空值的 KVPUB，以便在 TTL 过期时从所有客户端删除它。

可靠性

CHP 可用于双服务器配置，如果主服务器出现故障，备份服务器将接管。 CHP 没有指定用于此故障转移的机制，但二进制星型模式可能会有所帮助。

为了帮助服务器可靠性，客户端可以：

在每个 KVSET 命令中设置一个 UUID。
检测一段时间内 HUGZ 的缺失，并将其用作当前服务器出现故障的指标。
连接到备份服务器并重新请求状态同步。

可扩展性和性能

CHP 设计为可扩展到大量（数千）客户端，仅受代理上的系统资源限制。因为所有更新都通过单个服务器，所以总吞吐量将被限制在峰值时每秒数百万次更新，甚至可能更少。

安全

CHP 不实施任何身份验证、访问控制或加密机制，不应在需要这些的任何部署中使用。

构建多线程堆栈和 API
到目前为止，我们使用的客户端堆栈不够智能，无法正确处理此协议。一旦我们开始做心跳，我们就需要一个可以在后台线程中运行的客户端堆栈。在第 4 章末尾的自由模式 - 可靠的请求-回复模式中，我们使用了多线程 API，但没有详细解释。事实证明，当您开始制作更复杂的 ZeroMQ 协议（如 CHP）时，多线程 API 非常有用。

图 63 - 多线程 API

如果您制定了一个重要的协议，并且希望应用程序正确实现它，那么大多数开发人员在大多数情况下都会出错。你会留下很多不满的人抱怨你的协议太复杂、太脆弱、太难使用。而如果你给他们一个简单的 API 来调用，你就有机会让他们购买。

我们的多线程 API 由一个前端对象和一个后台代理组成，通过两个 PAIR 套接字连接。像这样连接两个 PAIR 套接字非常有用，以至于您的高级绑定可能应该执行 CZMQ 所做的工作，即打包一个“使用管道创建新线程，我可以使用它向其发送消息”方法。

我们在本书中看到的多线程 API 都采用相同的形式：

对象的构造函数 (clone_new) 创建上下文并启动与管道连接的后台线程。它固定在管道的一端，因此它可以向后台线程发送命令。
后台线程启动一个代理，它本质上是一个 zmq_poll 循环，从管道套接字和任何其他套接字（这里是 DEALER 和 SUB 套接字）读取。
主应用程序线程和后台线程现在仅通过 ZeroMQ 消息进行通信。按照惯例，前端发送字符串命令，以便类上的每个方法都变成发送给后端代理的消息，如下所示：

void
clone_connect (clone_t *self, char *address, char *service)
{assert (self);zmsg_t *msg = zmsg_new ();zmsg_addstr (msg, "CONNECT");zmsg_addstr (msg, address);zmsg_addstr (msg, service);zmsg_send (&msg, self->pipe);
}

如果该方法需要返回码，它可以等待来自代理的回复消息。
如果代理需要将异步事件发送回前端，我们向类添加一个 recv 方法，该方法等待前端管道上的消息。
我们可能希望公开前端管道套接字句柄以允许将类集成到进一步的轮询循环中。否则任何 recv 方法都会阻塞应用程序。

clone 类与第 4 章 - 可靠的请求-回复模式中的 flcliapi 类具有相同的结构，并添加了来自 Clone 客户端的最后一个模型的逻辑。如果没有 ZeroMQ，这种多线程 API 设计将需要数周的艰苦工作。使用 ZeroMQ，只需一两天的工作。

克隆类的实际 API 方法非常简单：

//  Create a new clone class instance
clone_t *clone_new (void);//  Destroy a clone class instance
voidclone_destroy (clone_t **self_p);//  Define the subtree, if any, for this clone class
voidclone_subtree (clone_t *self, char *subtree);//  Connect the clone class to one server
voidclone_connect (clone_t *self, char *address, char *service);//  Set a value in the shared hashmap
voidclone_set (clone_t *self, char *key, char *value, int ttl);//  Get a value from the shared hashmap
char *clone_get (clone_t *self, char *key);

所以这里是克隆客户端的模型六，它现在已经变成了一个使用克隆类的瘦壳：

//  Clone client Model Six//  Lets us build this source without creating a library
#include "clone.c"
#define SUBTREE "/client/"int main (void)
{//  Create distributed hash instanceclone_t *clone = clone_new ();//  Specify configurationclone_subtree (clone, SUBTREE);clone_connect (clone, "tcp://localhost", "5556");clone_connect (clone, "tcp://localhost", "5566");//  Set random tuples into the distributed hashwhile (!zctx_interrupted) {//  Set random value, check it was storedchar key [255];char value [10];sprintf (key, "%s%d", SUBTREE, randof (10000));sprintf (value, "%d", randof (1000000));clone_set (clone, key, value, randof (30));sleep (1);}clone_destroy (&clone);return 0;
}

请注意 connect 方法，该方法指定了一个服务器端点。在幕后，我们实际上正在与三个端口交谈。但是，正如 CHP 协议所说，三个端口位于连续的端口号上：

服务器状态路由器 (ROUTER) 位于端口 P。
服务器更新发布者 (PUB) 位于端口 P + 1。
服务器更新订阅者 (SUB) 位于端口 P + 2。

所以我们可以将三个连接折叠成一个逻辑操作（我们将其实现为三个单独的 ZeroMQ 连接调用）。

让我们以克隆堆栈的源代码结束。这是一段复杂的代码，但将其分解为前端对象类和后端代理时更容易理解。前端向代理发送字符串命令（“SUBTREE”、“CONNECT”、“SET”、“GET”），代理处理这些命令并与服务器通信。这是代理的逻辑：

通过从第一台服务器获取快照来启动
当我们获得快照时，切换到从订阅者套接字读取。
如果我们没有获得快照，则故障转移到第二台服务器。
轮询管道和订阅者套接字。
如果我们在管道上获得输入，则处理来自前端对象的控制消息。
如果我们获得了订阅者的输入，则存储或应用更新。
如果我们在一定时间内没有从服务器获得任何信息，请进行故障转移。
重复直到过程被 Ctrl-C 中断。

这是实际的克隆类实现：

clone: Clone class in C
//  clone class - Clone client API stack (multithreaded)#include "clone.h"
//  If no server replies within this time, abandon request
#define GLOBAL_TIMEOUT  4000    //  msecs//  =====================================================================
//  Synchronous part, works in our application thread//  Structure of our classstruct _clone_t {zctx_t *ctx;                //  Our context wrappervoid *pipe;                 //  Pipe through to clone agent
};//  This is the thread that handles our real clone class
static void clone_agent (void *args, zctx_t *ctx, void *pipe);//  .split constructor and destructor
//  Here are the constructor and destructor for the clone class. Note that
//  we create a context specifically for the pipe that connects our
//  frontend to the backend agent:clone_t *
clone_new (void)
{clone_t*self;self = (clone_t *) zmalloc (sizeof (clone_t));self->ctx = zctx_new ();self->pipe = zthread_fork (self->ctx, clone_agent, NULL);return self;
}void
clone_destroy (clone_t **self_p)
{assert (self_p);if (*self_p) {clone_t *self = *self_p;zctx_destroy (&self->ctx);free (self);*self_p = NULL;}
}//  .split subtree method
//  Specify subtree for snapshot and updates, which we must do before
//  connecting to a server as the subtree specification is sent as the
//  first command to the server. Sends a [SUBTREE][subtree] command to
//  the agent:void clone_subtree (clone_t *self, char *subtree)
{assert (self);zmsg_t *msg = zmsg_new ();zmsg_addstr (msg, "SUBTREE");zmsg_addstr (msg, subtree);zmsg_send (&msg, self->pipe);
}//  .split connect method
//  Connect to a new server endpoint. We can connect to at most two
//  servers. Sends [CONNECT][endpoint][service] to the agent:void
clone_connect (clone_t *self, char *address, char *service)
{assert (self);zmsg_t *msg = zmsg_new ();zmsg_addstr (msg, "CONNECT");zmsg_addstr (msg, address);zmsg_addstr (msg, service);zmsg_send (&msg, self->pipe);
}//  .split set method
//  Set a new value in the shared hashmap. Sends a [SET][key][value][ttl]
//  command through to the agent which does the actual work:void
clone_set (clone_t *self, char *key, char *value, int ttl)
{char ttlstr [10];sprintf (ttlstr, "%d", ttl);assert (self);zmsg_t *msg = zmsg_new ();zmsg_addstr (msg, "SET");zmsg_addstr (msg, key);zmsg_addstr (msg, value);zmsg_addstr (msg, ttlstr);zmsg_send (&msg, self->pipe);
}//  .split get method
//  Look up value in distributed hash table. Sends [GET][key] to the agent and
//  waits for a value response. If there is no value available, will eventually
//  return NULL:char *
clone_get (clone_t *self, char *key)
{assert (self);assert (key);zmsg_t *msg = zmsg_new ();zmsg_addstr (msg, "GET");zmsg_addstr (msg, key);zmsg_send (&msg, self->pipe);zmsg_t *reply = zmsg_recv (self->pipe);if (reply) {char *value = zmsg_popstr (reply);zmsg_destroy (&reply);return value;}return NULL;
}//  .split working with servers
//  The backend agent manages a set of servers, which we implement using
//  our simple class model:typedef struct {char *address;              //  Server addressint port;                   //  Server portvoid *snapshot;             //  Snapshot socketvoid *subscriber;           //  Incoming updatesuint64_t expiry;            //  When server expiresuint requests;              //  How many snapshot requests made?
} server_t;static server_t *
server_new (zctx_t *ctx, char *address, int port, char *subtree)
{server_t *self = (server_t *) zmalloc (sizeof (server_t));zclock_log ("I: adding server %s:%d...", address, port);self->address = strdup (address);self->port = port;self->snapshot = zsocket_new (ctx, ZMQ_DEALER);zsocket_connect (self->snapshot, "%s:%d", address, port);self->subscriber = zsocket_new (ctx, ZMQ_SUB);zsocket_connect (self->subscriber, "%s:%d", address, port + 1);zsocket_set_subscribe (self->subscriber, subtree);zsocket_set_subscribe (self->subscriber, "HUGZ");return self;
}static void
server_destroy (server_t **self_p)
{assert (self_p);if (*self_p) {server_t *self = *self_p;free (self->address);free (self);*self_p = NULL;}
}//  .split backend agent class
//  Here is the implementation of the backend agent itself://  Number of servers to which we will talk to
#define SERVER_MAX      2//  Server considered dead if silent for this long
#define SERVER_TTL      5000    //  msecs//  States we can be in
#define STATE_INITIAL       0   //  Before asking server for state
#define STATE_SYNCING       1   //  Getting state from server
#define STATE_ACTIVE        2   //  Getting new updates from servertypedef struct {zctx_t *ctx;                //  Context wrappervoid *pipe;                 //  Pipe back to applicationzhash_t *kvmap;             //  Actual key/value tablechar *subtree;              //  Subtree specification, if anyserver_t *server [SERVER_MAX];uint nbr_servers;           //  0 to SERVER_MAXuint state;                 //  Current stateuint cur_server;            //  If active, server 0 or 1int64_t sequence;           //  Last kvmsg processedvoid *publisher;            //  Outgoing updates
} agent_t;static agent_t *
agent_new (zctx_t *ctx, void *pipe)
{agent_t *self = (agent_t *) zmalloc (sizeof (agent_t));self->ctx = ctx;self->pipe = pipe;self->kvmap = zhash_new ();self->subtree = strdup ("");self->state = STATE_INITIAL;self->publisher = zsocket_new (self->ctx, ZMQ_PUB);return self;
}static void
agent_destroy (agent_t **self_p)
{assert (self_p);if (*self_p) {agent_t *self = *self_p;int server_nbr;for (server_nbr = 0; server_nbr < self->nbr_servers; server_nbr++)server_destroy (&self->server [server_nbr]);zhash_destroy (&self->kvmap);free (self->subtree);free (self);*self_p = NULL;}
}//  .split handling a control message
//  Here we handle the different control messages from the frontend;
//  SUBTREE, CONNECT, SET, and GET:static int
agent_control_message (agent_t *self)
{zmsg_t *msg = zmsg_recv (self->pipe);char *command = zmsg_popstr (msg);if (command == NULL)return -1;      //  Interruptedif (streq (command, "SUBTREE")) {free (self->subtree);self->subtree = zmsg_popstr (msg);}elseif (streq (command, "CONNECT")) {char *address = zmsg_popstr (msg);char *service = zmsg_popstr (msg);if (self->nbr_servers < SERVER_MAX) {self->server [self->nbr_servers++] = server_new (self->ctx, address, atoi (service), self->subtree);//  We broadcast updates to all known serverszsocket_connect (self->publisher, "%s:%d",address, atoi (service) + 2);}elsezclock_log ("E: too many servers (max. %d)", SERVER_MAX);free (address);free (service);}else//  .split set and get commands//  When we set a property, we push the new key-value pair onto//  all our connected servers:if (streq (command, "SET")) {char *key = zmsg_popstr (msg);char *value = zmsg_popstr (msg);char *ttl = zmsg_popstr (msg);//  Send key-value pair on to serverkvmsg_t *kvmsg = kvmsg_new (0);kvmsg_set_key  (kvmsg, key);kvmsg_set_uuid (kvmsg);kvmsg_fmt_body (kvmsg, "%s", value);kvmsg_set_prop (kvmsg, "ttl", ttl);kvmsg_send     (kvmsg, self->publisher);kvmsg_store    (&kvmsg, self->kvmap);free (key);free (value);free (ttl);}elseif (streq (command, "GET")) {char *key = zmsg_popstr (msg);kvmsg_t *kvmsg = (kvmsg_t *) zhash_lookup (self->kvmap, key);byte *value = kvmsg? kvmsg_body (kvmsg): NULL;if (value)zmq_send (self->pipe, value, kvmsg_size (kvmsg), 0);elsezstr_send (self->pipe, "");free (key);}free (command);zmsg_destroy (&msg);return 0;
}//  .split backend agent
//  The asynchronous agent manages a server pool and handles the
//  request-reply dialog when the application asks for it:static void
clone_agent (void *args, zctx_t *ctx, void *pipe)
{agent_t *self = agent_new (ctx, pipe);while (true) {zmq_pollitem_t poll_set [] = {{ pipe, 0, ZMQ_POLLIN, 0 },{ 0,    0, ZMQ_POLLIN, 0 }};int poll_timer = -1;int poll_size = 2;server_t *server = self->server [self->cur_server];switch (self->state) {case STATE_INITIAL://  In this state we ask the server for a snapshot,//  if we have a server to talk to...if (self->nbr_servers > 0) {zclock_log ("I: waiting for server at %s:%d...",server->address, server->port);if (server->requests < 2) {zstr_sendm (server->snapshot, "ICANHAZ?");zstr_send  (server->snapshot, self->subtree);server->requests++;}server->expiry = zclock_time () + SERVER_TTL;self->state = STATE_SYNCING;poll_set [1].socket = server->snapshot;}elsepoll_size = 1;break;case STATE_SYNCING://  In this state we read from snapshot and we expect//  the server to respond, else we fail over.poll_set [1].socket = server->snapshot;break;case STATE_ACTIVE://  In this state we read from subscriber and we expect//  the server to give HUGZ, else we fail over.poll_set [1].socket = server->subscriber;break;}if (server) {poll_timer = (server->expiry - zclock_time ())* ZMQ_POLL_MSEC;if (poll_timer < 0)poll_timer = 0;}//  .split client poll loop//  We're ready to process incoming messages; if nothing at all//  comes from our server within the timeout, that means the//  server is dead:int rc = zmq_poll (poll_set, poll_size, poll_timer);if (rc == -1)break;              //  Context has been shut downif (poll_set [0].revents & ZMQ_POLLIN) {if (agent_control_message (self))break;          //  Interrupted}elseif (poll_set [1].revents & ZMQ_POLLIN) {kvmsg_t *kvmsg = kvmsg_recv (poll_set [1].socket);if (!kvmsg)break;          //  Interrupted//  Anything from server resets its expiry timeserver->expiry = zclock_time () + SERVER_TTL;if (self->state == STATE_SYNCING) {//  Store in snapshot until we're finishedserver->requests = 0;if (streq (kvmsg_key (kvmsg), "KTHXBAI")) {self->sequence = kvmsg_sequence (kvmsg);self->state = STATE_ACTIVE;zclock_log ("I: received from %s:%d snapshot=%d",server->address, server->port,(int) self->sequence);kvmsg_destroy (&kvmsg);}elsekvmsg_store (&kvmsg, self->kvmap);}elseif (self->state == STATE_ACTIVE) {//  Discard out-of-sequence updates, incl. HUGZif (kvmsg_sequence (kvmsg) > self->sequence) {self->sequence = kvmsg_sequence (kvmsg);kvmsg_store (&kvmsg, self->kvmap);zclock_log ("I: received from %s:%d update=%d",server->address, server->port,(int) self->sequence);}elsekvmsg_destroy (&kvmsg);}}else {//  Server has died, failover to nextzclock_log ("I: server at %s:%d didn't give HUGZ",server->address, server->port);self->cur_server = (self->cur_server + 1) % self->nbr_servers;self->state = STATE_INITIAL;}}agent_destroy (&self);
}

zmq Pub-Sub 模式相关推荐

Redis学习笔记~分布式的Pub/Sub模式
redis的客户端有很多,这次用它的pub/sub发布与订阅我选择了StackExchange.Redis,发布与订阅大家应该很清楚了,首先一个订阅者,订阅一个服务,服务执行一些处理程序(可能是写个日 ...
ZMQ之克隆模式的可靠性
克隆服务器的可靠性克隆模型1至5相对比较简单,下面我们会探讨一个非常复杂的模型.可以发现,为了构建可靠的消息队列,我们需要花费非常多的精力.所以我们经常会问:有必要这么做吗?如果说你能够接受可靠性不 ...
Redis的Pub/Sub模式
为什么80%的码农都做不了架构师?>>> Redis同样支持消息的发布/订阅(Pub/Sub)模式,这和中间件activemq有些类似.订阅者(Subscriber)可以订阅自 ...
nanomsg(4) :pub/sub 模式
pub/sub(订阅发布)模式了解过mqtt的朋友们对这一套并不陌生.多个进程间通过订阅主题和发布主题,完成数据的交互. 需要注意的是,数据帧的格式: ${主题} | ${ 数据} // sub.c ...
使用python基于zmq的DEALER-ROUTER模式实现分布式消息分发的demo
zmq的DEALER套接字是对REQ的一层包装,本质就是在发送数据前发送一段bytes数据(所以一个bytes数据就意味着一个连接),而ROUTER就是对REP的包装,是在接收数据前先接收bytes数 ...
java zmq订阅_从ZMQ PUB套接字获取订户过滤器
使用 zmq.XPUB 套接字类型,有一种方法可以检测新用户和离开用户 . 以下代码示例显示了如何: # Publisher side import zmq ctx = zmq.Context.ins ...
EJB3.0学习笔记---MDBbean--区分P2P模式和Pub/Sub模式的示例
1.注意在写MDB程序的时候,如果是使用的P2P方式的话,可能你写了多个同样的MDBbean,但是只会有一个MDBbean被调用,至于是哪一个被调用是随机的,这时候其他的将不会起作用; 一个消息 ...
ZMQ发布-订阅模式踩坑之——消息丢失部分
如题最近在研究发布订阅者模式时,网上找的例子运行后没啥问题,但自己将其嵌入生产者消费者模式之后,出现了订阅端收到消息丢失很大一部分的问题.后逐渐排查,问题出在这里: zmq_send(publish ...
ZMQ特点及消息模式
ZMQ特点普通的socket是端对端的关系,ZMQ是N:M的关系,socket的连接需要显式地建立连接,销毁连接,选择协议(TCP/UDP)和错误处理,ZMQ屏蔽了这些细节,像是一个封装了的sock ...
zmq之pub/sub会出现丢包的问题的解决方案
1 应用场景当前要做的一个内容是通过好几个pub 客户端向一个 sub 服务器publish数据信息,sub服务器用于将所得到的信息解析存储到数据库,但是出现了中间丢掉一些数据包的问题,如下所示,我 ...

zmq Pub-Sub 模式

zmq Pub-Sub 模式相关推荐

最新文章

热门文章