ZMQ之克隆模式的可靠性
克隆服务器的可靠性
克隆模型1至5相对比较简单，下面我们会探讨一个非常复杂的模型。可以发现，为了构建可靠的消息队列，我们需要花费非常多的精力。所以我们经常会问：有必要这么做吗？如果说你能够接受可靠性不够高的、或者说已经足够好的架构，那恭喜你，你在成本和收益之间找到了平衡。虽然我们会偶尔丢失一些消息，但从经济的角度来说还是合理的。不管怎样，下面我们就来介绍这个复杂的模型。
在模型3中，你会关闭和重启服务，这会导致数据的丢失。任何后续加入的客户端只能得到重启之后的那些数据，而非所有的。下面就让我们想办法让克隆模式能够承担服务器重启的故障。
以下列举我们需要处理的问题：
1、克隆服务器进程崩溃并自动或手工重启。进程丢失了所有数据，所以必须从别处进行恢复。
2、克隆服务器硬件故障，长时间不能恢复。客户端需要切换至另一个可用的服务端。
3、克隆服务器从网络上断开，如交换机发生故障等。它会在某个时点重连，但期间的数据就需要替代的服务器负责处理。
第一步我们需要增加一个服务器。我们可以使用第四章中提到的双子星模式，它是一个反应堆，而我们的程序经过整理后也是一个反应堆，因此可以互相协作。
我们需要保证更新事件在主服务器崩溃时仍能保留，最简单的机制就是同时发送给两台服务器。
备机就可以当做一台客户端来运行，像其他客户端一样从主机获取更新事件。同时它又能从客户端获取更新事件——虽然不应该以此更新数据，但可以先暂存起来。
所以，相较于模型5，模型6中引入了以下特性：
1、客户端发送更新事件改用PUB-SUB套接字，而非PUSH-PULL。原因是PUSH套接字会在没有接收方时阻塞，且会进行负载均衡——我们需要两台服务器都接收到消息。我们会在服务器端绑定SUB套接字，在客户端连接PUB套接字。
2、我们在服务器发送给客户端的更新事件中加入心跳，这样客户端可以知道主机是否已死，然后切换至备机。
3、我们使用双子星模式的bstar反应堆类来创建主机和备机。双子星模式中需要有一个“投票”套接字，来协助判定对方节点是否已死。这里我们使用快照请求来作为“投票”。
4、我们将为所有的更新事件添加UUID属性，它由客户端生成，服务端会将其发布给所有客户端。
5、备机将维护一个“待处理列表”，保存来自客户端、尚未由服务端发布的更新事件；或者反过来，来自服务端、尚未从客户端收到的更新事件。这个列表从旧到新排列，这样就能方便地从顶部删除消息。
我们可以为客户端设计一个有限状态机，它有三种状态：
1、客户端打开并连接了套接字，然后向服务端发送快照请求。为了避免消息风暴，它只会请求两次。
2、客户端等待快照应答，如果获得了则保存它；如果没有获得，则向第二个服务器发送请求。
3、客户端收到快照，便开始等待更新事件。如果在一定时间内没有收到服务端响应，则会连接第二个服务端。
客户端会一直循环下去，可能在程序刚启动时，部分客户端会试图连接主机，部分连接备机，相信双子星模式会很好地处理这一情况的。
我们可以将客户端状态图绘制出来：
故障恢复的步骤如下：
1、客户端检测到主机不再发送心跳，因此转而连接备机，并请求一份新的快照；
2、备机开始接收快照请求，并检测到主机死亡，于是开始作为主机运行；
3、备机将待处理列表中的更新事件写入自身状态中，然后开始处理快照请求。
当主机恢复连接时：
1、启动为slave状态，并作为克隆模式客户端连接备机；
2、同时，使用SUB套接字从客户端接收更新事件。
我们做两点假设：
1、至少有一台主机会继续运行。如果两台主机都崩溃了，那我们将丢失所有的服务端数据，无法恢复。
2、不同的客户端不会同时更新同一个键值对。客户端的更新事件会先后到达两个服务器，因此更新的顺序可能会不一致。单个客户端的更新事件到达两台服务器的顺序是相同的，所以不用担心。
下面是整体架构图：
开始编程之前，我们需要将客户端重构成一个可复用的类。在ZMQ中写异步类有时是为了练习如何写出优雅的代码，但这里我们确实是希望克隆模式可以成为一种易于使用的程序。上述架构的伸缩性来源于客户端的正确行为，因此有必要将其封装成一份API。要在客户端中进行故障恢复还是比较复杂的，试想一下自由者模式和克隆模式结合起来会是什么样的吧。
按照我的习惯，我会先写出一份API的列表，然后加以实现。让我们假想一个名为clone的API，在其基础之上编写克隆模式客户端API。将代码封装为API显然会提升代码的稳定性，就以模型5为例，客户端需要打开三个套接字，端点名称直接写在了代码里。我们可以创建这样一组API：
// 为每个套接字指定端点 clone_subscribe (clone, "tcp://localhost:5556"); clone_snapshot (clone, "tcp://localhost:5557"); clone_updates (clone, "tcp://localhost:5558"); // 由于有两个服务端，因此再执行一次 clone_subscribe (clone, "tcp://localhost:5566"); clone_snapshot (clone, "tcp://localhost:5567"); clone_updates (clone, "tcp://localhost:5568");
但这种写法还是比较啰嗦的，因为没有必要将API内部的一些设计暴露给编程人员。现在我们会使用三个套接字，而将来可能就会使用两个，或者四个。我们不可能让所有的应用程序都相应地修改吧？让我们把这些信息包装到API中：
// 指定主备服务器端点 clone_connect (clone, "tcp://localhost:5551"); clone_connect (clone, "tcp://localhost:5561");
这样一来代码就变得非常简洁，不过也会对现有代码的内部就够造成影响。我们需要从一个端点中推算出三个端点。一种方法是假设客户端和服务端使用三个连续的端点通信，并将这个规则写入协议；另一个方法是向服务器索取缺少的端点信息。我们使用第一种较为简单的方法：
1、服务器状态ROUTER在端点P；
2、服务器更新事件PUB在端点P + 1；
3、服务器更新事件SUB在端点P + 2。
clone类和第四章的flcliapi类很类似，由两部分组成：
1、一个在后台运行的异步克隆模式代理。该代理处理所有的I/O操作，实时地和服务器进行通信；
2、一个在前台应用程序中同步运行的clone类。当你创建了一个clone对象后，它会自动创建后台的clone线程；当你销毁clone对象，该后台线程也会被销毁。
前台的clone类会使用inproc管道和后台的代理进行通信。C语言中，czmq线程会自动为我们创建这个管道。这也是ZMQ多线程编程的常规方式。
如果没有ZMQ，这种异步的设计将很难处理高压工作，而ZMQ会让其变得简单。编写出来额代码会相对比较复杂。我们可以用反应堆的模式来编写，但这会进一步增加复杂度，且影响应用程序的使用。因此，我们的设计的API将更像是一个能够和服务器进行通信的键值表：
clone_t *clone_new (void); void clone_destroy (clone_t **self_p); void clone_connect (clone_t *self, char *address, char *service); void clone_set (clone_t *self, char *key, char *value); char *clone_get (clone_t *self, char *key);
下面就是克隆模式客户端模型6的代码，因为调用了API，所以非常简短：
clonecli6: Clone client, Model Six in C
// // 克隆模式 - 客户端 - 模型6 // // 直接编译，不建类库 #include "clone.c" #define SUBTREE "/client/" int main (void) { // 创建分布式哈希表 clone_t *clone = clone_new (); // 配置 clone_subtree (clone, SUBTREE); clone_connect (clone, "tcp://localhost", "5556"); clone_connect (clone, "tcp://localhost", "5566"); // 插入随机键值 while (!zctx_interrupted) { // 生成随机值 char 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; }
以下是clone类的实现：
clone: Clone class in C
/* ===================================================================== clone - client-side Clone Pattern class --------------------------------------------------------------------- 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 "clone.h" // 请求超时时间 #define GLOBAL_TIMEOUT 4000 // msecs // 判定服务器死亡的时间 #define SERVER_TTL 5000 // msecs // 服务器数量 #define SERVER_MAX 2 // ===================================================================== // 同步部分，在应用程序线程中工作 // --------------------------------------------------------------------- // 类结构 struct _clone_t { zctx_t *ctx; // 上下文 void *pipe; // 和后台代理间的通信套接字 }; // 该线程用于处理真正的clone类 static void clone_agent (void *args, zctx_t *ctx, void *pipe); // --------------------------------------------------------------------- // 构造函数 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; } } // --------------------------------------------------------------------- // 在链接之前指定快照和更新事件的子树 // 发送给后台代理的消息内容为[SUBTREE][subtree] 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); } // --------------------------------------------------------------------- // 连接至新的服务器端点 // 消息内容：[CONNECT][endpoint][service] 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); } // --------------------------------------------------------------------- // 设置新值 // 消息内容：[SET][key][value][ttl] 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); } // --------------------------------------------------------------------- // 取值 // 消息内容：[GET][key] // 如果没有clone可用，会返回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; } // ===================================================================== // 异步部分，在后台运行 // --------------------------------------------------------------------- // 单个服务端信息 typedef struct { char *address; // 服务端地址 int port; // 端口 void *snapshot; // 快照套接字 void *subscriber; // 接收更新事件的套接字 uint64_t expiry; // 服务器过期时间 uint requests; // 收到的快照请求数 } 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); zsockopt_set_subscribe (self->subscriber, subtree); 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; } } // --------------------------------------------------------------------- // 后台代理类 // 状态 #define STATE_INITIAL 0 // 连接之前 #define STATE_SYNCING 1 // 正在同步 #define STATE_ACTIVE 2 // 正在更新 typedef struct { zctx_t *ctx; // 上下文 void *pipe; // 与主线程通信的套接字 zhash_t *kvmap; // 键值表 char *subtree; // 子树 server_t *server [SERVER_MAX]; uint nbr_servers; // 范围：0 - SERVER_MAX uint state; // 当前状态 uint cur_server; // 当前master，0/1 int64_t sequence; // 键值对编号 void *publisher; // 发布更新事件的套接字 } 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; } } // 若线程被中断则返回-1 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; if (streq (command, "SUBTREE")) { free (self->subtree); self->subtree = zmsg_popstr (msg); } else if (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); // 广播更新事件 zsocket_connect (self->publisher, "%s:%d", address, atoi (service) + 2); } else zclock_log ("E: too many servers (max. %d)", SERVER_MAX); free (address); free (service); } else if (streq (command, "SET")) { char *key = zmsg_popstr (msg); char *value = zmsg_popstr (msg); char *ttl = zmsg_popstr (msg); zhash_update (self->kvmap, key, (byte *) value); zhash_freefn (self->kvmap, key, free); // 向服务端发送键值对 kvmsg_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_destroy (&kvmsg); puts (key); free (ttl); free (key); // 键值对实际由哈希表对象控制 } else if (streq (command, "GET")) { char *key = zmsg_popstr (msg); char *value = zhash_lookup (self->kvmap, key); if (value) zstr_send (self->pipe, value); else zstr_send (self->pipe, ""); free (key); free (value); } free (command); zmsg_destroy (&msg); return 0; } // --------------------------------------------------------------------- // 异步的后台代理会维护一个服务端池，并处理来自应用程序的请求或应答。 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: // 该状态下，如果有可用服务，会发送快照请求 if (self->nbr_servers > 0) { zclock_log ("I: 正在等待服务器 %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; } else poll_size = 1; break; case STATE_SYNCING: // 该状态下我们从服务器端接收快照内容，若失败则尝试其他服务器 poll_set [1].socket = server->snapshot; break; case STATE_ACTIVE: // 该状态下我们从服务器获取更新事件，失败则尝试其他服务器 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; } // ------------------------------------------------------------ // poll循环 int rc = zmq_poll (poll_set, poll_size, poll_timer); if (rc == -1) break; // 上下文已被关闭 if (poll_set [0].revents & ZMQ_POLLIN) { if (agent_control_message (self)) break; // 中断 } else if (poll_set [1].revents & ZMQ_POLLIN) { kvmsg_t *kvmsg = kvmsg_recv (poll_set [1].socket); if (!kvmsg) break; // 中断 // 任何服务端的消息将重置它的过期时间 server->expiry = zclock_time () + SERVER_TTL; if (self->state == STATE_SYNCING) { // 保存快照内容 server->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); } else kvmsg_store (&kvmsg, self->kvmap); } else if (self->state == STATE_ACTIVE) { // 丢弃过期的更新事件 if (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); } else kvmsg_destroy (&kvmsg); } } else { // 服务端已死，尝试其他服务器 zclock_log ("I: 服务器 %s:%d 无响应", server->address, server->port); self->cur_server = (self->cur_server + 1) % self->nbr_servers; self->state = STATE_INITIAL; } } agent_destroy (&self); }
最后是克隆服务器的模型6代码：
clonesrv6: Clone server, Model Six in C
// // 克隆模式 - 服务端 - 模型6 // // 直接编译，不建类库 #include "bstar.c" #include "kvmsg.c" // bstar反应堆API 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); static int s_send_hugz (zloop_t *loop, void *socket, void *args); static int s_new_master (zloop_t *loop, void *unused, void *args); static int s_new_slave (zloop_t *loop, void *unused, void *args); static int s_subscriber (zloop_t *loop, void *socket, void *args); // 服务端属性 typedef struct { zctx_t *ctx; // 上下文 zhash_t *kvmap; // 存放键值对 bstar_t *bstar; // bstar反应堆核心 int64_t sequence; // 更新事件编号 int port; // 主端口 int peer; // 同伴端口 void *publisher; // 发布更新事件的端口 void *collector; // 接收客户端更新事件的端口 void *subscriber; // 接受同伴更新事件的端口 zlist_t *pending; // 延迟的更新事件 Bool primary; // 是否为主机 Bool master; // 是否为master Bool slave; // 是否为slave } clonesrv_t; 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: 作为主机master运行，正在等待备机slave连接。"); 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; } else if (argc == 2 && streq (argv [1], "-b")) { zclock_log ("I: 作为备机slave运行，正在等待主机master连接。"); 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: clonesrv4 { -p | -b }\n"); free (self); exit (0); } // 主机将成为master if (self->primary) self->kvmap = zhash_new (); self->ctx = zctx_new (); self->pending = zlist_new (); bstar_set_verbose (self->bstar, TRUE); // 设置克隆服务端套接字 self->publisher = zsocket_new (self->ctx, ZMQ_PUB); self->collector = zsocket_new (self->ctx, ZMQ_SUB); zsocket_bind (self->publisher, "tcp://*:%d", self->port + 1); zsocket_bind (self->collector, "tcp://*:%d", self->port + 2); // 作为克隆客户端连接同伴 self->subscriber = zsocket_new (self->ctx, ZMQ_SUB); zsocket_connect (self->subscriber, "tcp://localhost:%d", self->peer + 1); // 注册状态事件处理器 bstar_new_master (self->bstar, s_new_master, self); bstar_new_slave (self->bstar, s_new_slave, self); // 注册bstar反应堆其他事件处理器 zloop_reader (bstar_zloop (self->bstar), self->collector, 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); // 开启bstar反应堆 bstar_start (self->bstar); // 中断，终止。 while (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; } // --------------------------------------------------------------------- // 发送快照内容 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; } // --------------------------------------------------------------------- // 从客户端收集更新事件 // 如果我们是master，则将该事件写入kvmap对象； // 如果我们是slave，则将其写入延迟队列 static int s_was_pending (clonesrv_t *self, kvmsg_t *kvmsg); static int s_collector (zloop_t *loop, void *collector, void *args) { clonesrv_t *self = (clonesrv_t *) args; kvmsg_t *kvmsg = kvmsg_recv (collector); kvmsg_dump (kvmsg); if (kvmsg) { if (self->master) { 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); } else { // 如果我们已经从master中获得了该事件，则丢弃该消息 if (s_was_pending (self, kvmsg)) kvmsg_destroy (&kvmsg); else zlist_append (self->pending, kvmsg); } } return 0; } // 如果消息已在延迟队列中，则删除它并返回TRUE 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_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; } // --------------------------------------------------------------------- // 发送心跳 static int s_send_hugz (zloop_t *loop, void *unused, 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; } // --------------------------------------------------------------------- // 状态改变事件处理函数 // 我们将转变为master // // 备机先将延迟列表中的事件更新到自己的快照中， // 并开始接收客户端发来的快照请求。 static int s_new_master (zloop_t *loop, void *unused, void *args) { clonesrv_t *self = (clonesrv_t *) args; self->master = TRUE; self->slave = FALSE; zloop_cancel (bstar_zloop (self->bstar), self->subscriber); // 应用延迟列表中的事件 while (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: 正在发布延迟列表中的更新事件：%d", (int) self->sequence); } return 0; } // --------------------------------------------------------------------- // 正在切换为slave static int s_new_slave (zloop_t *loop, void *unused, void *args) { clonesrv_t *self = (clonesrv_t *) args; zhash_destroy (&self->kvmap); self->master = FALSE; self->slave = TRUE; zloop_reader (bstar_zloop (self->bstar), self->subscriber, s_subscriber, self); return 0; } // --------------------------------------------------------------------- // 从同伴主机（master）接收更新事件； // 接收该类更新事件时，我们一定是slave。 static int s_subscriber (zloop_t *loop, void *subscriber, void *args) { clonesrv_t *self = (clonesrv_t *) args; // 获取快照，如果需要的话。 if (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: 正在请求快照：tcp://localhost:%d", self->peer); zstr_send (snapshot, "ICANHAZ?"); while (TRUE) { kvmsg_t *kvmsg = kvmsg_recv (snapshot); if (!kvmsg) break; // 中断 if (streq (kvmsg_key (kvmsg), "KTHXBAI")) { self->sequence = kvmsg_sequence (kvmsg); kvmsg_destroy (&kvmsg); break; // 完成 } kvmsg_store (&kvmsg, self->kvmap); } zclock_log ("I: 收到快照，版本号：%d", (int) self->sequence); zsocket_destroy (self->ctx, snapshot); } // 查找并删除 kvmsg_t *kvmsg = kvmsg_recv (subscriber); if (!kvmsg) return 0; if (strneq (kvmsg_key (kvmsg), "HUGZ")) { if (!s_was_pending (self, kvmsg)) { // 如果master的更新事件比客户端的事件早到，则将master的事件存入延迟列表， // 当收到客户端更新事件时会将其从列表中清除。 zlist_append (self->pending, kvmsg_dup (kvmsg)); } // 如果更新事件比kvmap版本高，则应用它 if (kvmsg_sequence (kvmsg) > self->sequence) { self->sequence = kvmsg_sequence (kvmsg); kvmsg_store (&kvmsg, self->kvmap); zclock_log ("I: 收到更新事件：%d", (int) self->sequence); } else kvmsg_destroy (&kvmsg); } else kvmsg_destroy (&kvmsg); return 0; }
这段程序只有几百行，但还是花了一些时间来进行调通的。这个模型中包含了故障恢复，瞬间值，子树等等。虽然我们前期设计得很完备，但要在多个套接字之间进行调试还是很困难的。以下是我的工作方式：
1、由于使用了反应堆（bstar，建立在zloop之上），我们节省了大量的代码，让程序变得简洁明了。整个服务以一个线程运行，因此不会出现跨线程的问题。只需将结构指针（self）传递给所有的处理器即可。此外，使用发应堆后可以让代码更为模块化，易于重用。
2、我们逐个模块进行调试，只有某个模块能够正常运行时才会进入下一步。由于使用了四五个套接字，因此调试的工作量是很大的。我直接将调试信息输出到了屏幕上，因为实在没有必要专门开一个调试器来工作。
3、因为一直在使用valgrind工具进行测试，因此我能确定程序没有内存泄漏的问题。在C语言中，内存泄漏是我们非常关心的问题，因为没有什么垃圾回收机制可以帮你完成。正确地使用像kvmsg、czmq之类的抽象层可以很好地避免内存泄漏。
这段程序肯定还会存在一些BUG，部分读者可能会帮助我调试和修复，我在此表示感谢。
测试模型6时，先开启主机和备机，再打开一组客户端，顺序随意。随机地中止某个服务进程，如果程序设计得是正确的，那客户端获得的数据应该都是一致的。
