ZMQ之共享键值缓存（克隆模式）
发布-订阅模式和无线电广播有些类似，在你收听之前发送的消息你将无从得知，收到消息的多少又会取决于你的接收能力。让人吃惊的是，对于那些追求完美的工程师来说，这种机器恰恰符合他们的需求，且广为传播，成为现实生活中分发消息的最佳机制。想想非死不可、推特、BBS新闻、体育新闻等应用就知道了。
但是，在很多情形下，可靠的发布-订阅模式同样是有价值的。正如我们讨论请求-应答模式一样，我们会根据“故障”来定义“可靠性”，下面几项便是发布-订阅模式中可能发生的故障：
1、订阅者连接太慢，因此没有收到发布者最初发送的消息；
2、订阅者速度太慢，同样会丢失消息；
3、订阅者可能会断开，其间的消息也会丢失。
还有一些情况我们碰到的比较少，但不是没有：
1、订阅者崩溃、重启，从而丢失了所有已收到的消息；
2、订阅者处理消息的速度过慢，导致消息在队列中堆砌并溢出；
3、因网络过载而丢失消息（特别是PGM协议下的连接）；
4、网速过慢，消息在发布者处溢出，从而崩溃。
其实还会有其他出错的情况，只是以上这些在现实应用中是比较典型的。
我们已经有方法解决上面的某些问题了，比如对于慢速订阅者可以使用自杀的蜗牛模式。但是，对于其他的问题，我们最后能有一个可复用的框架来编写可靠的发布-订阅模式。
难点在于，我们并不知道目标应用程序会怎样处理这些数据。它们会进行过滤、只处理一部分消息吗？它们是否会将消息记录起来供日后使用？它们是否会将消息转发给其下的worker进行处理？需要考虑的情况实在太多了，每种情况都有其所谓的可靠性。
所以，我们将问题抽象出来，供多种应用程序使用。这种抽象应用我们称之为共享的键值缓存，它的功能是通过唯一的键名存储二进制数据块。
不要将这个抽象应用和分布式哈希表混淆起来，它是用来解决节点在分布式网络中相连接的问题的；也不要和分布式键值表混淆，它更像是一个NoSQL数据库。我们要建立的应用是将内存中的状态可靠地传递给一组客户端，它要做到的是：
1、客户端可以随时加入网络，并获得服务端当前的状态；
2、任何客户端都可以改变键值缓存（插入、更新、删除）；
3、将这种变化以最短的延迟可靠地传达给所有的客户端；
4、能够处理大量的客户端，成百上千。
克隆模式的要点在于客户端会反过来和服务端进行通信，这在简单的发布-订阅模式中并不常见。所以我这里使用“服务端”、“客户端”而不是“发布者”、“订阅者”这两个词。我们会使用发布-订阅模式作为核心消息模式，不过还需要夹杂其他模式。
分发键值更新事件
我们会分阶段实施克隆模式。首先，我们看看如何从服务器发送键值更新事件给所有的客户端。我们将第一章中使用的天气服务模型进行改造，以键值对的方式发送信息，并让客户端使用哈希表来保存：
以下是服务端代码：
clonesrv1: Clone server, Model One in C
// // 克隆模式服务端模型1 // // 让我们直接编译，不生成类库 #include "kvsimple.c" int main (void) { // 准备上下文和PUB套接字 zctx_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) { // 使用键值对分发消息 kvmsg_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 (" 已中止\n已发送 %d 条消息\n", (int) sequence); zhash_destroy (&kvmap); zctx_destroy (&ctx); return 0; }
以下是客户端代码：
clonecli1: Clone client, Model One in C
// // 克隆模式客户端模型1 // // 让我们直接编译，不生成类库 #include "kvsimple.c" int main (void) { // 准备上下文和SUB套接字 zctx_t *ctx = zctx_new (); void *updates = zsocket_new (ctx, ZMQ_SUB); 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; // 中断 kvmsg_store (&kvmsg, kvmap); sequence++; } printf (" 已中断\n收到 %d 条消息\n", (int) sequence); zhash_destroy (&kvmap); zctx_destroy (&ctx); return 0; }
几点说明：
1、所有复杂的工作都在kvmsg类中完成了，这个类能够处理键值对类型的消息对象，其实质上是一个ZMQ多帧消息，共有三帧：键（ZMQ字符串）、编号（64位，按字节顺序排列）、二进制体（保存所有附加信息）。
2、服务端随机生成消息，使用四位数作为键，这样可以模拟大量而不是过量的哈希表（1万个条目）。
3、服务端绑定套接字后会等待200毫秒，以避免订阅者连接延迟而丢失数据的问题。我们会在后面的模型中解决这一点。
4、我们使用“发布者”和“订阅者”来命名程序中使用的套接字，这样可以避免和后续模型中的其他套接字发生混淆。
以下是kvmsg的代码，已经经过了精简：
kvsimple: Key-value message class in C
/* ===================================================================== kvsimple - simple key-value message class for example applications --------------------------------------------------------------------- Copyright (c) 1991-2011 iMatix Corporation <www.imatix.com> Copyright other contributors as noted in the AUTHORS file. This file is part of the ZeroMQ Guide: http://zguide.zeromq.org This is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published by the Free Software Foundation; either version 3 of the License, or (at your option) any later version. This software is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details. You should have received a copy of the GNU Lesser General Public License along with this program. If not, see <http://www.gnu.org/licenses/>. ===================================================================== */ #include "kvsimple.h" #include "zlist.h" // 键是一个短字符串 #define KVMSG_KEY_MAX 255 // 消息被格式化成三帧 // frame 0: 键（ZMQ字符串） // frame 1: 编号（8个字节，按顺序排列） // frame 2: 内容（二进制数据块） #define FRAME_KEY 0 #define FRAME_SEQ 1 #define FRAME_BODY 2 #define KVMSG_FRAMES 3 // 类结构 struct _kvmsg { // 消息中某帧是否存在 int present [KVMSG_FRAMES]; // 对应的ZMQ消息帧 zmq_msg_t frame [KVMSG_FRAMES]; // 将键转换为C语言字符串 char key [KVMSG_KEY_MAX + 1]; }; // --------------------------------------------------------------------- // 构造函数，设置编号 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_freefn()函数调用 void kvmsg_free (void *ptr) { if (ptr) { kvmsg_t *self = (kvmsg_t *) ptr; // 销毁消息中的帧 int 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 (self); } } void kvmsg_destroy (kvmsg_t **self_p) { assert (self_p); if (*self_p) { kvmsg_free (*self_p); *self_p = NULL; } } // --------------------------------------------------------------------- // 从套接字中读取键值消息，返回kvmsg实例 kvmsg_t * kvmsg_recv (void *socket) { assert (socket); kvmsg_t *self = kvmsg_new (0); // 读取所有帧，出错则销毁对象 int 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_recvmsg (socket, &self->frame [frame_nbr], 0) == -1) { kvmsg_destroy (&self); break; } // 验证多帧消息 int rcvmore = (frame_nbr < KVMSG_FRAMES - 1)? 1: 0; if (zsockopt_rcvmore (socket) != rcvmore) { kvmsg_destroy (&self); break; } } return self; } // --------------------------------------------------------------------- // 向套接字发送键值对消息，不检验消息帧的内容 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 (©); if (self->present [frame_nbr]) zmq_msg_copy (©, &self->frame [frame_nbr]); zmq_sendmsg (socket, ©, (frame_nbr < KVMSG_FRAMES - 1)? ZMQ_SNDMORE: 0); zmq_msg_close (©); } } // --------------------------------------------------------------------- // 从消息中获取键值，不存在则返回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; } else return NULL; } // --------------------------------------------------------------------- // 返回消息的编号 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; } else return 0; } // --------------------------------------------------------------------- // 返回消息内容，不存在则返回NULL byte * kvmsg_body (kvmsg_t *self) { assert (self); if (self->present [FRAME_BODY]) return (byte *) zmq_msg_data (&self->frame [FRAME_BODY]); else return NULL; } // --------------------------------------------------------------------- // 返回消息内容的大小 size_t kvmsg_size (kvmsg_t *self) { assert (self); if (self->present [FRAME_BODY]) return zmq_msg_size (&self->frame [FRAME_BODY]); else return 0; } // --------------------------------------------------------------------- // 设置消息的键 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_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; } // --------------------------------------------------------------------- // 设置消息内容 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); } // --------------------------------------------------------------------- // 使用printf()格式设置消息键 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); } // --------------------------------------------------------------------- // 使用springf()格式设置消息内容 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)); } // --------------------------------------------------------------------- // 若kvmsg结构的键值均存在，则存入哈希表； // 如果kvmsg结构已没有引用，则自动销毁和释放。 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; } } // --------------------------------------------------------------------- // 将消息内容打印至标准错误输出，用以调试和跟踪 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"); } else fprintf (stderr, "NULL message\n"); } // --------------------------------------------------------------------- // 测试用例 int kvmsg_test (int verbose) { kvmsg_t *kvmsg; printf (" * kvmsg: "); // 准备上下文和套接字 zctx_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 (); // 测试简单消息的发送和接受 kvmsg = 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); // 关闭并销毁所有对象 zhash_destroy (&kvmap); zctx_destroy (&ctx); printf ("OK\n"); return 0; }
我们会在下文编写一个更为完整的kvmsg类，可以用到现实环境中。
客户端和服务端都会维护一个哈希表，但这个模型需要所有的客户端都比服务端启动得早，而且不能崩溃，这显然不能满足可靠性的要求。
创建快照
为了让后续连接的（或从故障中恢复的）客户端能够获取服务器上的状态信息，需要让它在连接时获取一份快照。正如我们将“消息”的概念简化为“已编号的键值对”，我们也可以将“状态”简化为“一个哈希表”。为获取服务端状态，客户端会打开一个REQ套接字进行请求：
我们需要考虑时间的问题，因为生成快照是需要一定时间的，我们需要知道应从哪个更新事件开始更新快照，服务端是不知道何时有更新事件的。一种方法是先开始订阅消息，收到第一个消息之后向服务端请求“将该条更新之前的所有内容发送给”。这样一来，服务器需要为每一次更新保存一份快照，这显然是不现实的。
所以，我们会在客户端用以下方式进行同步：
1、客户端开始订阅服务器的更新事件，然后请求一份快照。这样就能保证这份快照是在上一次更新事件之后产生的。
2、客户端开始等待服务器的快照，并将更新事件保存在队列中，做法很简单，不要从套接字中读取消息就可以了，ZMQ会自动将这些消息保存起来，这时不应设置阈值（HWM）。
3、当客户端获取到快照后，它将再次开始读取更新事件，但是需要丢弃那些早于快照生成时间的事件。如快照生成时包含了200次更新，那客户端会从第201次更新开始读取。
4、随后，客户端就会用更新事件去更新自身的状态了。
这是一个比较简单的模型，因为它用到了ZMQ消息队列的机制。服务端代码如下：
clonesrv2: Clone server, Model Two in C
// // 克隆模式 - 服务端 - 模型2 // // 让我们直接编译，不创建类库 #include "kvsimple.c" static int s_send_single (char *key, void *data, void *args); static void state_manager (void *args, zctx_t *ctx, void *pipe); int main (void) { // 准备套接字和上下文 zctx_t *ctx = zctx_new (); void *publisher = zsocket_new (ctx, ZMQ_PUB); zsocket_bind (publisher, "tcp://*:5557"); int64_t sequence = 0; srandom ((unsigned) time (NULL)); // 开启状态管理器，并等待同步信号 void *updates = zthread_fork (ctx, state_manager, NULL); free (zstr_recv (updates)); while (!zctx_interrupted) { // 分发键值消息 kvmsg_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 (" 已中断\n已发送 %d 条消息\n", (int) sequence); zctx_destroy (&ctx); return 0; } // 快照请求方信息 typedef struct { void *socket; // 用于发送快照的ROUTER套接字 zframe_t *identity; // 请求方的标识 } kvroute_t; // 发送快照中单个键值对 // 使用kvmsg对象作为载体 static int s_send_single (char *key, void *data, void *args) { kvroute_t *kvroute = (kvroute_t *) args; // 先发送接收方标识 zframe_send (&kvroute->identity, kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE); kvmsg_t *kvmsg = (kvmsg_t *) data; kvmsg_send (kvmsg, kvroute->socket); return 0; } // 该线程维护服务端状态，并处理快照请求。 // 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; // 当前快照版本 while (!zctx_interrupted) { int rc = zmq_poll (items, 2, -1); if (rc == -1 && errno == ETERM) break; // 上下文异常 // 等待主线程的更新事件 if (items [0].revents & ZMQ_POLLIN) { kvmsg_t *kvmsg = kvmsg_recv (pipe); if (!kvmsg) break; // 中断 sequence = kvmsg_sequence (kvmsg); kvmsg_store (&kvmsg, kvmap); } // 执行快照请求 if (items [1].revents & ZMQ_POLLIN) { zframe_t *identity = zframe_recv (snapshot); if (!identity) break; // 中断 // 请求内容在第二帧中 char *request = zstr_recv (snapshot); if (streq (request, "ICANHAZ?")) free (request); else { printf ("E: 错误的请求，程序中止\n"); break; } // 发送快照给客户端 kvroute_t routing = { snapshot, identity }; // 逐项发送 zhash_foreach (kvmap, s_send_single, &routing); // 发送结束标识，内含快照版本号 printf ("正在发送快照，版本号 %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); }
以下是客户端代码：
clonecli2: Clone client, Model Two in C
// // 克隆模式 - 客户端 - 模型2 // // 让我们直接编译，不生成类库 #include "kvsimple.c" int main (void) { // 准备上下文和SUB套接字 zctx_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_connect (subscriber, "tcp://localhost:5557"); zhash_t *kvmap = zhash_new (); // 获取快照 int64_t sequence = 0; zstr_send (snapshot, "ICANHAZ?"); while (TRUE) { kvmsg_t *kvmsg = kvmsg_recv (snapshot); if (!kvmsg) break; // 中断 if (streq (kvmsg_key (kvmsg), "KTHXBAI")) { sequence = kvmsg_sequence (kvmsg); printf ("已获取快照，版本号=%d\n", (int) sequence); kvmsg_destroy (&kvmsg); break; // 完成 } kvmsg_store (&kvmsg, kvmap); } // 应用队列中的更新事件，丢弃过时事件 while (!zctx_interrupted) { kvmsg_t *kvmsg = kvmsg_recv (subscriber); if (!kvmsg) break; // 中断 if (kvmsg_sequence (kvmsg) > sequence) { sequence = kvmsg_sequence (kvmsg); kvmsg_store (&kvmsg, kvmap); } else kvmsg_destroy (&kvmsg); } zhash_destroy (&kvmap); zctx_destroy (&ctx); return 0; }
几点说明：
1、客户端使用两个线程，一个用来生成随机的更新事件，另一个用来管理状态。两者之间使用PAIR套接字通信。可能你会考虑使用SUB套接字，但是“慢连接”的问题会影响到程序运行。PAIR套接字会让两个线程严格同步的。
2、我们在updates套接字上设置了阈值（HWM），避免更新服务内存溢出。在inproc协议的连接中，阈值是两端套接字阈值的加和，所以要分别设置。
3、客户端比较简单，用C语言编写，大约60行代码。大多数工作都在kvmsg类中完成了，不过总的来说，克隆模式实现起来还是比较简单的。
4、我们没有用特别的方式来序列化状态内容。键值对用kvmsg对象表示，保存在一个哈希表中。在不同的时间请求状态时会得到不同的快照。
5、我们假设客户端只和一个服务进行通信，而且服务必须是正常运行的。我们暂不考虑如何从服务崩溃的情形中恢复过来。
现在，这两段程序都还没有真正地工作起来，但已经能够正确地同步状态了。这是一个多种消息模式的混合体：进程内的PAIR、发布-订阅、ROUTER-DEALER等。
重发键值更新事件
第二个模型中，键值更新事件都来自于服务器，构成了一个中心化的模型。但是我们需要的是一个能够在客户端进行更新的缓存，并能同步到其他客户端中。这时，服务端只是一个无状态的中间件，带来的好处有：
1、我们不用太过关心服务端的可靠性，因为即使它崩溃了，我们仍能从客户端中获取完整的数据。
2、我们可以使用键值缓存在动态节点之间分享数据。
客户端的键值更新事件会通过PUSH-PULL套接字传达给服务端：
我们为什么不让客户端直接将更新信息发送给其他客户端呢？虽然这样做可以减少延迟，但是就无法为更新事件添加自增的唯一编号了。很多应用程序都需要更新事件以某种方式排序，只有将消息发给服务端，由服务端分发更新消息，才能保证更新事件的顺序。
有了唯一的编号后，客户端还能检测到更多的故障：网络堵塞或队列溢出。如果客户端发现消息输入流有一段空白，它能采取措施。可能你会觉得此时让客户端通知服务端，让它重新发送丢失的信息，可以解决问题。但事实上没有必要这么做。消息流的空挡表示网络状况不好，如果再进行这样的请求，只会让事情变得更糟。所以一般的做法是由客户端发出警告，并停止运行，等到有专人来维护后再继续工作。
我们开始创建在客户端进行状态更新的模型。以下是客户端代码：
clonesrv3: Clone server, Model Three in C
// // 克隆模式 服务端 模型3 // // 直接编译，不创建类库 #include "kvsimple.c" static int s_send_single (char *key, void *data, void *args); // 快照请求方信息 typedef struct { void *socket; // ROUTER套接字 zframe_t *identity; // 请求方标识 } kvroute_t; int main (void) { // 准备上下文和套接字 zctx_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); // 执行来自客户端的更新事件 if (items [0].revents & ZMQ_POLLIN) { kvmsg_t *kvmsg = kvmsg_recv (collector); if (!kvmsg) break; // 中断 kvmsg_set_sequence (kvmsg, ++sequence); kvmsg_send (kvmsg, publisher); kvmsg_store (&kvmsg, kvmap); printf ("I: 发布更新事件 %5d\n", (int) sequence); } // 响应快照请求 if (items [1].revents & ZMQ_POLLIN) { zframe_t *identity = zframe_recv (snapshot); if (!identity) break; // 中断 // 请求内容在消息的第二帧中 char *request = zstr_recv (snapshot); if (streq (request, "ICANHAZ?")) free (request); else { printf ("E: 错误的请求，程序中止\n"); break; } // 发送快照 kvroute_t routing = { snapshot, identity }; // 逐条发送 zhash_foreach (kvmap, s_send_single, &routing); // 发送结束标识和编号 printf ("I: 正在发送快照，版本号：%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 (" 已中断\n已处理 %d 条消息\n", (int) sequence); zhash_destroy (&kvmap); zctx_destroy (&ctx); return 0; } // 发送一条键值对状态给套接字，使用kvmsg对象保存键值对 static int s_send_single (char *key, void *data, void *args) { kvroute_t *kvroute = (kvroute_t *) args; // Send identity of recipient first zframe_send (&kvroute->identity, kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE); kvmsg_t *kvmsg = (kvmsg_t *) data; kvmsg_send (kvmsg, kvroute->socket); return 0; }
以下是客户端代码：
clonecli3: Clone client, Model Three in C
// // 克隆模式 - 客户端 - 模型3 // // 直接编译，不创建类库 #include "kvsimple.c" int main (void) { // 准备上下文和SUB套接字 zctx_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_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)); // 获取状态快照 int64_t sequence = 0; zstr_send (snapshot, "ICANHAZ?"); while (TRUE) { kvmsg_t *kvmsg = kvmsg_recv (snapshot); if (!kvmsg) break; // 中断 if (streq (kvmsg_key (kvmsg), "KTHXBAI")) { sequence = kvmsg_sequence (kvmsg); printf ("I: 已收到快照，版本号：%d\n", (int) sequence); kvmsg_destroy (&kvmsg); break; // 完成 } 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; // 上下文被关闭 if (items [0].revents & ZMQ_POLLIN) { kvmsg_t *kvmsg = kvmsg_recv (subscriber); if (!kvmsg) break; // 中断 // 丢弃过时消息，包括心跳 if (kvmsg_sequence (kvmsg) > sequence) { sequence = kvmsg_sequence (kvmsg); kvmsg_store (&kvmsg, kvmap); printf ("I: 收到更新事件：%d\n", (int) sequence); } else kvmsg_destroy (&kvmsg); } // 创建一个随机的更新事件 if (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 (" 已准备\n收到 %d 条消息\n", (int) sequence); zhash_destroy (&kvmap); zctx_destroy (&ctx); return 0; }
几点说明：
1、服务端整合为一个线程，负责收集来自客户端的更新事件并转发给其他客户端。它使用PULL套接字获取更新事件，ROUTER套接字处理快照请求，以及PUB套接字发布更新事件。
2、客户端会每隔1秒左右发送随机的更新事件给服务端，现实中这一动作由应用程序触发。
子树克隆
现实中的键值缓存会越变越多，而客户端可能只会需要部分缓存。我们可以使用子树的方式来实现：客户端在发送快照请求时告诉服务端它需要的子树，在订阅更新事件时也指明子树。
关于子树的语法有很多，一种是“分层路径”结构，另一种是“主题树”
1、分层路径：/some/list/of/paths。
2、主题树：some.list.of.topics
这里我们会使用分层路径结构，以此扩展服务端和客户端，进行子树操作。维护多个子树其实并不太困难，因此我们不在这里演示。
下面是服务端代码，由模型3衍化而来：
clonesrv4: Clone server, Model Four in C
// // 克隆模式 服务端 模型4 // // 直接编译，不创建类库 #include "kvsimple.c" static int s_send_single (char *key, void *data, void *args); // 快照请求方信息 typedef struct { void *socket; // ROUTER套接字 zframe_t *identity; // 请求方标识 char *subtree; // 指定的子树 } kvroute_t; int main (void) { // 准备上下文和套接字 zctx_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); // 执行来自客户端的更新事件 if (items [0].revents & ZMQ_POLLIN) { kvmsg_t *kvmsg = kvmsg_recv (collector); if (!kvmsg) break; // Interrupted kvmsg_set_sequence (kvmsg, ++sequence); kvmsg_send (kvmsg, publisher); kvmsg_store (&kvmsg, kvmap); printf ("I: 发布更新事件 %5d\n", (int) sequence); } // 响应快照请求 if (items [1].revents & ZMQ_POLLIN) { zframe_t *identity = zframe_recv (snapshot); if (!identity) break; // Interrupted // 请求内容在消息的第二帧中 char *request = zstr_recv (snapshot); char *subtree = NULL; if (streq (request, "ICANHAZ?")) { free (request); subtree = zstr_recv (snapshot); } else { printf ("E: 错误的请求，程序中止\n"); break; } // 发送快照 kvroute_t routing = { snapshot, identity, subtree }; // 逐条发送 zhash_foreach (kvmap, s_send_single, &routing); // 发送结束标识和编号 printf ("I: 正在发送快照，版本号：%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); } } printf (" 已中断\n已处理 %d 条消息\n", (int) sequence); zhash_destroy (&kvmap); zctx_destroy (&ctx); return 0; } // 发送一条键值对状态给套接字，使用kvmsg对象保存键值对 static int s_send_single (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, kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE); kvmsg_send (kvmsg, kvroute->socket); } return 0; }
下面是客户端代码：
clonecli4: Clone client, Model Four in C
// // 克隆模式 - 客户端 - 模型4 // // 直接编译，不创建类库 #include "kvsimple.c" #define SUBTREE "/client/" int main (void) { // 准备上下文和SUB套接字 zctx_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_connect (subscriber, "tcp://localhost:5557"); zsockopt_set_subscribe (subscriber, SUBTREE); void *publisher = zsocket_new (ctx, ZMQ_PUSH); zsocket_connect (publisher, "tcp://localhost:5558"); zhash_t *kvmap = zhash_new (); srandom ((unsigned) time (NULL)); // 获取状态快照 int64_t sequence = 0; zstr_sendm (snapshot, "ICANHAZ?"); zstr_send (snapshot, SUBTREE); while (TRUE) { kvmsg_t *kvmsg = kvmsg_recv (snapshot); if (!kvmsg) break; // Interrupted if (streq (kvmsg_key (kvmsg), "KTHXBAI")) { sequence = kvmsg_sequence (kvmsg); printf ("I: 已收到快照，版本号：%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; // 上下文被关闭 if (items [0].revents & ZMQ_POLLIN) { kvmsg_t *kvmsg = kvmsg_recv (subscriber); if (!kvmsg) break; // 中断 // 丢弃过时消息，包括心跳 if (kvmsg_sequence (kvmsg) > sequence) { sequence = kvmsg_sequence (kvmsg); kvmsg_store (&kvmsg, kvmap); printf ("I: 收到更新事件：%d\n", (int) sequence); } else kvmsg_destroy (&kvmsg); } // 创建一个随机的更新事件 if (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; } } printf (" 已准备\n收到 %d 条消息\n", (int) sequence); zhash_destroy (&kvmap); zctx_destroy (&ctx); return 0; }
瞬间值
瞬间值指的是那些会立刻过期的值。如果你用克隆模式搭建一个类似DNS的服务时，就可以用瞬间值来模拟动态DNS解析了。当节点连接网络，对外发布它的地址，并不断地更新地址。如果节点断开连接，则它的地址也会失效。
瞬间值可以和会话（session）联系起来，当会话结束时，瞬间值也就失效了。克隆模式中，会话是由客户端定义的，并会在客户端断开连接时消亡。
更简单的方法是为每一个瞬间值设定一个过期时间，客户端会不断延长这个时间，当断开连接时这个时间将得不到更新，服务器就会自动将其删除。
我们会用这种简单的方法来实现瞬间值，因为太过复杂的方法可能不值当，它们的差别仅在性能上体现。如果客户端有很多瞬间值，那为每个值设定过期时间是恰当的；如果瞬间值到达一定的量，那最好还是将其和会话相关联，统一进行过期处理。
首先，我们需要设法在键值对消息中加入过期时间。我们可以增加一个消息帧，但这样一来每当我们需要增加消息内容时就需要修改kvmsg类库了，这并不合适。所以，我们一次性增加一个“属性”消息帧，用于添加不同的消息属性。
其次，我们需要设法删除这条数据。目前为止服务端和客户端会盲目地增改哈希表中的数据，我们可以这样定义：当消息的值是空的，则表示删除这个键的数据。
下面是一个更为完整的kvmsg类代码，它实现了“属性”帧，以及一个UUID帧，我们后面会用到。该类还会负责处理值为空的消息，达到删除的目的：
kvmsg: Key-value message class - full in C
/* ===================================================================== kvmsg - key-value message class for example applications --------------------------------------------------------------------- Copyright (c) 1991-2011 iMatix Corporation <www.imatix.com> Copyright other contributors as noted in the AUTHORS file. This file is part of the ZeroMQ Guide: http://zguide.zeromq.org This is free software; you can redistribute it and/or modify it under the terms of the GNU Lesser General Public License as published by the Free Software Foundation; either version 3 of the License, or (at your option) any later version. This software is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details. You should have received a copy of the GNU Lesser General Public License along with this program. If not, see <http://www.gnu.org/licenses/>. ===================================================================== */ #include "kvmsg.h" #include <uuid/uuid.h> #include "zlist.h" // 键是短字符串 #define KVMSG_KEY_MAX 255 // 消息包含五帧 // frame 0: 键(ZMQ字符串) // frame 1: 编号(8个字节，按顺序排列) // frame 2: UUID(二进制块，16个字节) // frame 3: 属性(ZMQ字符串) // frame 4: 值(二进制块) #define FRAME_KEY 0 #define FRAME_SEQ 1 #define FRAME_UUID 2 #define FRAME_PROPS 3 #define FRAME_BODY 4 #define KVMSG_FRAMES 5 // 类结构 struct _kvmsg { // 帧是否存在 int present [KVMSG_FRAMES]; // 对应消息帧 zmq_msg_t frame [KVMSG_FRAMES]; // 键，C语言字符串格式 char key [KVMSG_KEY_MAX + 1]; // 属性列表，key=value形式 zlist_t *props; size_t props_size; }; // 将属性列表序列化为字符串 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); } } // --------------------------------------------------------------------- // 构造函数，指定消息编号 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()调用 void kvmsg_free (void *ptr) { if (ptr) { kvmsg_t *self = (kvmsg_t *) ptr; // 释放所有消息帧 int frame_nbr; for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++) if (self->present [frame_nbr]) zmq_msg_close (&self->frame [frame_nbr]); // 释放属性列表 while (zlist_size (self->props)) free (zlist_pop (self->props)); zlist_destroy (&self->props); // 释放对象本身 free (self); } } void kvmsg_destroy (kvmsg_t **self_p) { assert (self_p); if (*self_p) { kvmsg_free (*self_p); *self_p = NULL; } } // --------------------------------------------------------------------- // 复制kvmsg对象 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 = zlist_copy (self->props); return kvmsg; } // --------------------------------------------------------------------- // 从套接字总读取键值对，返回kvmsg实例 kvmsg_t * kvmsg_recv (void *socket) { assert (socket); kvmsg_t *self = kvmsg_new (0); // 读取所有帧，若有异常则直接返回空 int 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_recvmsg (socket, &self->frame [frame_nbr], 0) == -1) { kvmsg_destroy (&self); break; } // 验证多帧消息 int rcvmore = (frame_nbr < KVMSG_FRAMES - 1)? 1: 0; if (zsockopt_rcvmore (socket) != rcvmore) { kvmsg_destroy (&self); break; } } if (self) s_decode_props (self); return self; } // --------------------------------------------------------------------- // 向套接字发送键值对消息，空消息也发送 void kvmsg_send (kvmsg_t *self, void *socket) { assert (self); assert (socket); s_encode_props (self); int frame_nbr; for (frame_nbr = 0; frame_nbr < KVMSG_FRAMES; frame_nbr++) { zmq_msg_t copy; zmq_msg_init (©); if (self->present [frame_nbr]) zmq_msg_copy (©, &self->frame [frame_nbr]); zmq_sendmsg (socket, ©, (frame_nbr < KVMSG_FRAMES - 1)? ZMQ_SNDMORE: 0); zmq_msg_close (©); } } // --------------------------------------------------------------------- // 返回消息的键 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; } else return NULL; } // --------------------------------------------------------------------- // 返回消息的编号 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; } else return 0; } // --------------------------------------------------------------------- // 返回消息的UUID 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]); else return NULL; } // --------------------------------------------------------------------- // 返回消息的内容 byte * kvmsg_body (kvmsg_t *self) { assert (self); if (self->present [FRAME_BODY]) return (byte *) zmq_msg_data (&self->frame [FRAME_BODY]); else return NULL; } // --------------------------------------------------------------------- // 返回消息内容的长度 size_t kvmsg_size (kvmsg_t *self) { assert (self); if (self->present [FRAME_BODY]) return zmq_msg_size (&self->frame [FRAME_BODY]); else return 0; } // --------------------------------------------------------------------- // 设置消息的键 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_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; } // --------------------------------------------------------------------- // 生成并设置消息的UUID 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; } // --------------------------------------------------------------------- // 设置消息的内容 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); } // --------------------------------------------------------------------- // 使用printf()格式设置消息的键 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); } // --------------------------------------------------------------------- // 使用printf()格式设置消息内容 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)); } // --------------------------------------------------------------------- // 获取消息属性，无则返回空字符串 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 ""; } // --------------------------------------------------------------------- // 设置消息属性 // 属性名称不能包含=号，值的最大长度是255 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); // 分配空间 char *prop = malloc (strlen (name) + strlen (value) + 2); // 删除已存在的属性 sprintf (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); } // 添加新属性 strcat (prop, value); zlist_append (self->props, prop); self->props_size += strlen (prop) + 1; } // --------------------------------------------------------------------- // 在哈希表中保存kvmsg对象 // 当kvmsg对象不再被使用时进行释放操作； // 若传入的值为空，则删除该对象。 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); } } else zhash_delete (hash, kvmsg_key (self)); *self_p = NULL; } } // --------------------------------------------------------------------- // 将消息内容输出到标准错误输出 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); 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, "]"); } int char_nbr; for (char_nbr = 0; char_nbr < size; char_nbr++) fprintf (stderr, "%02X", body [char_nbr]); fprintf (stderr, "\n"); } else fprintf (stderr, "NULL message\n"); } // --------------------------------------------------------------------- // 测试用例 int kvmsg_test (int verbose) { kvmsg_t *kvmsg; printf (" * kvmsg: "); // 准备上下文和套接字 zctx_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 (); // 测试简单消息的收发 kvmsg = 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); // 测试带有属性的消息的收发 kvmsg = 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); // 关闭并销毁所有对象 zhash_destroy (&kvmap); zctx_destroy (&ctx); printf ("OK\n"); return 0; }
客户端模型5和模型4没有太大区别，只是kvmsg类库变了。在更新消息的时候还需要添加一个过期时间的属性：
kvmsg_set_prop (kvmsg, "ttl", "%d", randof (30));
服务端模型5有较大的变化，我们会用反应堆来代替轮询，这样就能混合处理定时事件和套接字事件了，只是在C语言中是比较麻烦的。下面是代码：
clonesrv5: Clone server, Model Five in C
// // 克隆模式 - 服务端 - 模型5 // // 直接编译，不建类库 #include "kvmsg.c" // 反应堆处理器 static int s_snapshots (zloop_t *loop, void *socket, void *args); static int s_collector (zloop_t *loop, void *socket, void *args); static int s_flush_ttl (zloop_t *loop, void *socket, void *args); // 服务器属性 typedef struct { zctx_t *ctx; // 上下文 zhash_t *kvmap; // 键值对存储 zloop_t *loop; // zloop反应堆 int port; // 主端口 int64_t sequence; // 更新事件编号 void *snapshot; // 处理快照请求 void *publisher; // 发布更新事件 void *collector; // 从客户端收集接收更新事件 } 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); // 打开克隆模式服务端套接字 self->snapshot = zsocket_new (self->ctx, ZMQ_ROUTER); self->publisher = zsocket_new (self->ctx, ZMQ_PUB); self->collector = zsocket_new (self->ctx, ZMQ_PULL); zsocket_bind (self->snapshot, "tcp://*:%d", self->port); zsocket_bind (self->publisher, "tcp://*:%d", self->port + 1); zsocket_bind (self->collector, "tcp://*:%d", self->port + 2); // 注册反应堆处理程序 zloop_reader (self->loop, self->snapshot, s_snapshots, self); zloop_reader (self->loop, self->collector, s_collector, self); zloop_timer (self->loop, 1000, 0, s_flush_ttl, self); // 运行反应堆，直至中断 zloop_start (self->loop); zloop_destroy (&self->loop); zhash_destroy (&self->kvmap); zctx_destroy (&self->ctx); free (self); return 0; } // --------------------------------------------------------------------- // 发送快照内容 static int s_send_single (char *key, void *data, void *args); // 请求方信息 typedef struct { void *socket; // ROUTER套接字 zframe_t *identity; // 请求方标识 char *subtree; // 子树信息 } kvroute_t; static int s_snapshots (zloop_t *loop, void *snapshot, void *args) { clonesrv_t *self = (clonesrv_t *) args; zframe_t *identity = zframe_recv (snapshot); if (identity) { // 请求位于消息第二帧 char *request = zstr_recv (snapshot); char *subtree = NULL; if (streq (request, "ICANHAZ?")) { free (request); subtree = zstr_recv (snapshot); } else printf ("E: 错误的请求，程序中止\n"); if (subtree) { // 发送状态快照 kvroute_t routing = { snapshot, identity, subtree }; zhash_foreach (self->kvmap, s_send_single, &routing); // 发送结束符和版本号 zclock_log ("I: 正在发送快照，版本号：%d", (int) self->sequence); zframe_send (&identity, snapshot, ZFRAME_MORE); kvmsg_t *kvmsg = kvmsg_new (self->sequence); kvmsg_set_key (kvmsg, "KTHXBAI"); kvmsg_set_body (kvmsg, (byte *) subtree, 0); kvmsg_send (kvmsg, snapshot); kvmsg_destroy (&kvmsg); free (subtree); } } return 0; } // 每次发送一个快照键值对 static int s_send_single (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, kvroute->socket, ZFRAME_MORE + ZFRAME_REUSE); kvmsg_send (kvmsg, kvroute->socket); } return 0; } // --------------------------------------------------------------------- // 收集更新事件 static int s_collector (zloop_t *loop, void *collector, void *args) { clonesrv_t *self = (clonesrv_t *) args; kvmsg_t *kvmsg = kvmsg_recv (collector); 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: 正在发布更新事件 %d", (int) self->sequence); } return 0; } // --------------------------------------------------------------------- // 删除过期的瞬间值 static int s_flush_single (char *key, void *data, void *args); static int s_flush_ttl (zloop_t *loop, void *unused, void *args) { clonesrv_t *self = (clonesrv_t *) args; zhash_foreach (self->kvmap, s_flush_single, args); return 0; } // 删除过期的键值对，并广播该事件 static int s_flush_single (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: 发布删除事件 %d", (int) self->sequence); } return 0; }
