首页 > 代码库 > 高可靠对称节点(双星模式)
高可靠对称节点(双星模式)
高可靠对称节点(双星模式)
概览
双星模式是一对具有主从机制的高可靠节点。任一时间,某个节点会充当主机,接收所有客户端的请求;另一个则作为一种备机存在。两个节点会互相监控对方,当主机从网络中消失时,备机会替代主机的位置。
双星模式由Pieter Hintjens和Martin Sustrik设计,应用在iMatix的OpenAMQ服务器中。它的设计理念是:
- 提供一种简明的高可靠性解决方案;
- 易于理解和使用;
- 能够进行可靠的故障切换。
假设我们有一组双星模式的服务器,以下是可能发生的故障:
- 主机发生硬件故障(断电、失火等),应用程序发送后立刻使用备机进行连接;
- 主机的网络环境发生故障,可能某个路由器被雷击了,立刻使用备机;
- 主机上的服务被维护人员误杀,无法自动恢复。
恢复步骤如下:
- 维护人员排查主机故障;
- 将备机关闭,造成短时间的服务不可用;
- 待应用程序都连接到主机后,维护人员重启备机。
恢复过程是人工进行的,惨痛的经验告诉我们自动恢复是很可怕的:
故障的发生会造成10-30秒之间的服务暂停,如果这是一个真正的突发状况,那最好还是让主机暂停服务的好,因为立刻重启服务可能造成另一个10-30秒的暂停,不如让用户停止使用。
当有紧急状况发生时,可以在修复的过程中记录故障发生原因,而不是让系统自动恢复,管理员因此无法用其经验抵御下一次突发状况。
最后,如果自动恢复确实成功了,管理员将无从得知故障的发生原因,因而无法进行分析。
双星模式的故障恢复过程是:在修复了主机的问题后,将备机做关闭处理,稍后再重新开启:
双星模式的关闭过程有两种:
- 先关闭备机,等待一段时间后再关闭主机;
- 同时关闭主机和备机,间隔时间不超过几秒。
关闭时,间隔时间要比故障切换时间短,否则会导致应用程序失去连接、重新连接、并再次失去连接,导致用户投诉。
详细要求
双星模式可以非常简单,但能工作得很出色。事实上,这里的实现方法已经历经三个版本了,之前的版本都过于复杂,想要做太多的事情,因而被我们抛弃。我们需要的只是最基本的功能,能够提供易理解、易开发、高可靠的解决方法就可以了。
以下是该架构的详细需求:
- 需要用到双星模式的故障是:系统遭受灾难性的打击,如硬件崩溃、火灾、意外等。对于其他常规的服务器故障,可以用更简单的方法。
- 故障恢复时间应该在60秒以内,理想情况下应该在10秒以内;
- 故障恢复(failover)应该是自动完成的,而系统还原(recover)则是由人工完成的。我们希望应用程序能够在发生故障时自动从主机切换到备机,但不希望在问题解决之前自动切换回主机,因为这很有可能让主机再次崩溃。
- 程序的逻辑应该尽量简单,易于使用,最好能封装在API中;
- 需要提供一个明确的指示,哪台主机正在提供服务,以避免“精神分裂”的症状,即两台服务器都认为自己是主机;
- 两台服务器的启动顺序不应该有限制;
- 启动或关闭主从机时不需要更改客户端的配置,但有可能会中断连接;
- 管理员需要能够同时监控两台机器;
- 两台机器之间必须有专用的高速网络连接,必须能使用特定IP进行路由。
我们做如下架假设:
- 单台备机能够提供足够的保障,不需要再进行其他备份机制;
- 主从机应该都能够提供完整的服务,承载相同的压力,不需要进行负载均衡;
- 预算中允许有这样一台长时间闲置的备机。
双星模式不会用到:
- 多台备机,或在主从机之间进行负载均衡。该模式中的备机将一直处于空闲状态,只有主机发生问题时才会工作;
- 处理持久化的消息或事务。我们假设所连接的网络是不可靠的(或不可信的)。
- 自动搜索网络。双星模式是手工配置的,他们知道对方的存在,应用程序则知道双星的存在。
- 主从机之间状态的同步。所有服务端的状态必须能由应用程序进行重建。
以下是双星模式中的几个术语:
- 主机 - 通常情况下作为master的机器;
- 备机 - 通常情况下作为slave的机器,只有当主机从网络中消失时,备机才会切换成master状态,接收所有的应用程序请求;
- master - 双星模式中接收应用程序请求的机器;同一时刻只有一台master;
- slave - 当master消失时用以顶替的机器。
配置双星模式的步骤:
- 让主机知道备机的位置;
- 让备机知道主机的位置;
- 调整故障恢复时间,两台机器的配置必须相同。
比较重要的配置是应让两台机器间隔多久检查一次对方的状态,以及多长时间后采取行动。在我们的示例中,故障恢复时间设置为2000毫秒,超过这个时间备机就会代替主机的位置。但若你将主机的服务包裹在一个shell脚本中进行重启,就需要延长这个时间,否则备机可能在主机恢复连接的过程中转换成master。
要让客户端应用程序和双星模式配合,你需要做的是:
- 知道两台服务器的地址;
- 尝试连接主机,若失败则连接备机;
- 检测失效的连接,一般使用心跳机制;
- 尝试重连主机,然后再连接备机,其间的间隔应比服务器故障恢复时间长;
- 重建服务器端需要的所有状态数据;
- 如果要保证可靠性,应重发故障期间的消息。
这不是件容易的事,所以我们一般会将其封装成一个API,供程序员使用。
双星模式的主要限制有:
- 服务端进程不能涉及到一个以上的双星对称节点;
- 主机只能有一个备机;
- 当备机处于slave状态时,它不会处理任何请求;
- 备机必须能够承受所有的应用程序请求;
- 故障恢复时间不能在运行时调整;
- 客户端应用程序需要做一些重连的工作。
防止精神分裂
“精神分裂”症状指的是一个集群中的不同部分同时认为自己是master,从而停止对对方的检测。双星模式中的算法会降低这种症状的发生几率:主备机在决定自己是否为master时会检测自身是否收到了应用程序的请求,以及对方是否已经从网络中消失。
但在某些情况下,双星模式也会发生精神分裂。比如说,主备机被配置在两幢大楼里,每幢大楼的局域网中又分布了一些应用程序。这样,当两幢大楼的网络通信被阻断,双星模式的主备机就会分别在两幢大楼里接受和处理请求。
为了防止精神分裂,我们必须让主备机使用专用的网络进行连接,最简单的方法当然是用一根双绞线将他们相连。
我们不能将双星部署在两个不同的岛屿上,为各自岛屿的应用程序服务。这种情况下,我们会使用诸如联邦模式的机制进行可靠性设计。
最好但最夸张的做法是,将两台机器之间的连接和应用程序的连接完全隔离开来,甚至是使用不同的网卡,而不仅仅是不同的端口。这样做也是为了日后排查错误时更为明确。
实现双星模式
闲话少说,下面是双星模式的服务端代码:
bstarsrv: Binary Star server in C
//// 双星模式 - 服务端//#include "czmq.h"// 发送状态信息的间隔时间// 如果对方在两次心跳过后都没有应答,则视为断开#define HEARTBEAT 1000 // In msecs// 服务器状态枚举typedef enum { STATE_PRIMARY = 1, // 主机,等待同伴连接 STATE_BACKUP = 2, // 备机,等待同伴连接 STATE_ACTIVE = 3, // 激活态,处理应用程序请求 STATE_PASSIVE = 4 // 被动态,不接收请求} state_t;// 对话节点事件typedef enum { PEER_PRIMARY = 1, // 主机 PEER_BACKUP = 2, // 备机 PEER_ACTIVE = 3, // 激活态 PEER_PASSIVE = 4, // 被动态 CLIENT_REQUEST = 5 // 客户端请求} event_t;// 有限状态机typedef struct { state_t state; // 当前状态 event_t event; // 当前事件 int64_t peer_expiry; // 判定节点死亡的时限} bstar_t;// 执行有限状态机(将事件绑定至状态);// 发生异常时返回TRUE。static Bools_state_machine (bstar_t *fsm){ Bool exception = FALSE; // 主机等待同伴连接 // 该状态下接收CLIENT_REQUEST事件 if (fsm->state == STATE_PRIMARY) { if (fsm->event == PEER_BACKUP) { printf ("I: 已连接至备机(slave),可以作为master运行。\n"); fsm->state = STATE_ACTIVE; } else if (fsm->event == PEER_ACTIVE) { printf ("I: 已连接至备机(master),可以作为slave运行。\n"); fsm->state = STATE_PASSIVE; } } else // 备机等待同伴连接 // 该状态下拒绝CLIENT_REQUEST事件 if (fsm->state == STATE_BACKUP) { if (fsm->event == PEER_ACTIVE) { printf ("I: 已连接至主机(master),可以作为slave运行。\n"); fsm->state = STATE_PASSIVE; } else if (fsm->event == CLIENT_REQUEST) exception = TRUE; } else // 服务器处于激活态 // 该状态下接受CLIENT_REQUEST事件 if (fsm->state == STATE_ACTIVE) { if (fsm->event == PEER_ACTIVE) { // 若出现两台master,则抛出异常 printf ("E: 严重错误:双master。正在退出。\n"); exception = TRUE; } } else // 服务器处于被动态 // 若同伴已死,CLIENT_REQUEST事件将触发故障恢复 if (fsm->state == STATE_PASSIVE) { if (fsm->event == PEER_PRIMARY) { // 同伴正在重启 - 转为激活态,同伴将转为被动态。 printf ("I: 主机(slave)正在重启,可作为master运行。\n"); fsm->state = STATE_ACTIVE; } else if (fsm->event == PEER_BACKUP) { // 同伴正在重启 - 转为激活态,同伴将转为被动态。 printf ("I: 备机(slave)正在重启,可作为master运行。\n"); fsm->state = STATE_ACTIVE; } else if (fsm->event == PEER_PASSIVE) { // 若出现两台slave,集群将无响应 printf ("E: 严重错误:双slave。正在退出\n"); exception = TRUE; } else if (fsm->event == CLIENT_REQUEST) { // 若心跳超时,同伴将成为master; // 此行为由客户端请求触发。 assert (fsm->peer_expiry > 0); if (zclock_time () >= fsm->peer_expiry) { // 同伴已死,转为激活态。 printf ("I: 故障恢复,可作为master运行。\n"); fsm->state = STATE_ACTIVE; } else // 同伴还在,拒绝请求。 exception = TRUE; } } return exception;}int main (int argc, char *argv []){ // 命令行参数可以为: // -p 作为主机启动, at tcp://localhost:5001 // -b 作为备机启动, at tcp://localhost:5002 zctx_t *ctx = zctx_new (); void *statepub = zsocket_new (ctx, ZMQ_PUB); void *statesub = zsocket_new (ctx, ZMQ_SUB); void *frontend = zsocket_new (ctx, ZMQ_ROUTER); bstar_t fsm = { 0 }; if (argc == 2 && streq (argv [1], "-p")) { printf ("I: 主机master,等待备机(slave)连接。\n"); zsocket_bind (frontend, "tcp://*:5001"); zsocket_bind (statepub, "tcp://*:5003"); zsocket_connect (statesub, "tcp://localhost:5004"); fsm.state = STATE_PRIMARY; } else if (argc == 2 && streq (argv [1], "-b")) { printf ("I: 备机slave,等待主机(master)连接。\n"); zsocket_bind (frontend, "tcp://*:5002"); zsocket_bind (statepub, "tcp://*:5004"); zsocket_connect (statesub, "tcp://localhost:5003"); fsm.state = STATE_BACKUP; } else { printf ("Usage: bstarsrv { -p | -b }\n"); zctx_destroy (&ctx); exit (0); } // 设定下一次发送状态的时间 int64_t send_state_at = zclock_time () + HEARTBEAT; while (!zctx_interrupted) { zmq_pollitem_t items [] = { { frontend, 0, ZMQ_POLLIN, 0 }, { statesub, 0, ZMQ_POLLIN, 0 } }; int time_left = (int) ((send_state_at - zclock_time ())); if (time_left < 0) time_left = 0; int rc = zmq_poll (items, 2, time_left * ZMQ_POLL_MSEC); if (rc == -1) break; // 上下文对象被关闭 if (items [0].revents & ZMQ_POLLIN) { // 收到客户端请求 zmsg_t *msg = zmsg_recv (frontend); fsm.event = CLIENT_REQUEST; if (s_state_machine (&fsm) == FALSE) // 返回应答 zmsg_send (&msg, frontend); else zmsg_destroy (&msg); } if (items [1].revents & ZMQ_POLLIN) { // 收到状态消息,作为事件处理 char *message = zstr_recv (statesub); fsm.event = atoi (message); free (message); if (s_state_machine (&fsm)) break; // 错误,退出。 fsm.peer_expiry = zclock_time () + 2 * HEARTBEAT; } // 定时发送状态信息 if (zclock_time () >= send_state_at) { char message [2]; sprintf (message, "%d", fsm.state); zstr_send (statepub, message); send_state_at = zclock_time () + HEARTBEAT; } } if (zctx_interrupted) printf ("W: 中断\n"); // 关闭套接字和上下文 zctx_destroy (&ctx); return 0;}
下面是客户端代码:
bstarcli: Binary Star client in C
//// 双星模式 - 客户端//#include "czmq.h"#define REQUEST_TIMEOUT 1000 // 毫秒#define SETTLE_DELAY 2000 // 超时时间int main (void){ zctx_t *ctx = zctx_new (); char *server [] = { "tcp://localhost:5001", "tcp://localhost:5002" }; uint server_nbr = 0; printf ("I: 正在连接服务器 %s...\n", server [server_nbr]); void *client = zsocket_new (ctx, ZMQ_REQ); zsocket_connect (client, server [server_nbr]); int sequence = 0; while (!zctx_interrupted) { // 发送请求并等待应答 char request [10]; sprintf (request, "%d", ++sequence); zstr_send (client, request); int expect_reply = 1; while (expect_reply) { // 轮询套接字 zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } }; int rc = zmq_poll (items, 1, REQUEST_TIMEOUT * ZMQ_POLL_MSEC); if (rc == -1) break; // 中断 // 处理应答 if (items [0].revents & ZMQ_POLLIN) { // 审核应答编号 char *reply = zstr_recv (client); if (atoi (reply) == sequence) { printf ("I: 服务端应答正常 (%s)\n", reply); expect_reply = 0; sleep (1); // 每秒发送一个请求 } else { printf ("E: 错误的应答内容: %s\n", reply); } free (reply); } else { printf ("W: 服务器无响应,正在重试\n"); // 重开套接字 zsocket_destroy (ctx, client); server_nbr = (server_nbr + 1) % 2; zclock_sleep (SETTLE_DELAY); printf ("I: 正在连接服务端 %s...\n", server [server_nbr]); client = zsocket_new (ctx, ZMQ_REQ); zsocket_connect (client, server [server_nbr]); // 使用新套接字重发请求 zstr_send (client, request); } } } zctx_destroy (&ctx); return 0;}
运行以下命令进行测试,顺序随意:
bstarsrv -p # Start primarybstarsrv -b # Start backupbstarcli
可以将主机进程杀掉,测试故障恢复机制;再开启主机,杀掉备机,查看还原机制。要注意是由客户端触发这两个事件的。
下图展现了服务进程的状态图。绿色状态下会接收客户端请求,粉色状态会拒绝请求。事件指的是同伴的状态,所以“同伴激活态”指的是同伴机器告知我们它处于激活态。“客户请求”表示我们从客户端获得了请求,“客户投票”则指我们从客户端获得了请求并且同伴已经超时死亡。
需要注意的是,服务进程使用PUB-SUB套接字进行状态交换,其它类型的套接字在这里不适用。比如,PUSH和DEALER套接字在没有节点相连的时候会发生阻塞;PAIR套接字不会在节点断开后进行重连;ROUTER套接字需要地址才能发送消息。
These are the main limitations of the Binary Star pattern:
- A server process cannot be part of more than one Binary Star pair.
- A primary server can have a single backup server, no more.
- The backup server cannot do useful work while in slave mode.
- The backup server must be capable of handling full application loads.
- Failover configuration cannot be modified at runtime.
- Client applications must do some work to benefit from failover.
双星反应堆
我们可以将双星模式打包成一个类似反应堆的类,供以后复用。在C语言中,我们使用czmq的zloop类,其他语言应该会有相应的实现。以下是C语言版的bstar接口:
// 创建双星模式实例,使用本地(绑定)和远程(连接)端点来设置节点对。bstar_t *bstar_new (int primary, char *local, char *remote);// 销毁实例void bstar_destroy (bstar_t **self_p);// 返回底层的zloop反应堆,用以添加定时器、读取器、注册和取消等功能。zloop_t *bstar_zloop (bstar_t *self);// 注册投票读取器int bstar_voter (bstar_t *self, char *endpoint, int type,zloop_fn handler, void *arg);// 注册状态机处理器void bstar_new_master (bstar_t *self, zloop_fn handler, void *arg);void bstar_new_slave (bstar_t *self, zloop_fn handler, void *arg);// 开启反应堆,当回调函数返回-1,或进程收到SIGINT、SIGTERM信号时中止。int bstar_start (bstar_t *self);
以下是类的实现:
bstar: Binary Star core class in C
/* ===================================================================== bstar - Binary Star reactor --------------------------------------------------------------------- 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 "bstar.h"// 服务器状态枚举typedef enum { STATE_PRIMARY = 1, // 主机,等待同伴连接 STATE_BACKUP = 2, // 备机,等待同伴连接 STATE_ACTIVE = 3, // 激活态,处理应用程序请求 STATE_PASSIVE = 4 // 被动态,不接收请求} state_t;// 对话节点事件typedef enum { PEER_PRIMARY = 1, // 主机 PEER_BACKUP = 2, // 备机 PEER_ACTIVE = 3, // 激活态 PEER_PASSIVE = 4, // 被动态 CLIENT_REQUEST = 5 // 客户端请求} event_t;// 发送状态信息的间隔时间// 如果对方在两次心跳过后都没有应答,则视为断开#define BSTAR_HEARTBEAT 1000 // In msecs// 类结构struct _bstar_t { zctx_t *ctx; // 私有上下文 zloop_t *loop; // 反应堆循环 void *statepub; // 状态发布者 void *statesub; // 状态订阅者 state_t state; // 当前状态 event_t event; // 当前事件 int64_t peer_expiry; // 判定节点死亡的时限 zloop_fn *voter_fn; // 投票套接字处理器 void *voter_arg; // 投票处理程序的参数 zloop_fn *master_fn; // 成为master时回调 void *master_arg; // 参数 zloop_fn *slave_fn; // 成为slave时回调 void *slave_arg; // 参数};// ---------------------------------------------------------------------// 执行有限状态机(将事件绑定至状态);// 发生异常时返回-1,正确时返回0。static ints_execute_fsm (bstar_t *self){ int rc = 0; // 主机等待同伴连接 // 该状态下接收CLIENT_REQUEST事件 if (self->state == STATE_PRIMARY) { if (self->event == PEER_BACKUP) { zclock_log ("I: 已连接至备机(slave),可以作为master运行。"); self->state = STATE_ACTIVE; if (self->master_fn) (self->master_fn) (self->loop, NULL, self->master_arg); } else if (self->event == PEER_ACTIVE) { zclock_log ("I: 已连接至备机(master),可以作为slave运行。"); self->state = STATE_PASSIVE; if (self->slave_fn) (self->