一、引言
我们讲了那么多关于中间件的示例，好像有些违背“ZMQ是无中间件”的说法。但要知道在现实生活中，中间件一直是让人又爱又恨的东西。实践中的很多消息架构能都在使用中间件进行分布式架构的搭建，所以说最终的决定还是需要你自己去权衡的。这也是为什么虽然我能驾车10分钟到一个大型商场里购买五箱音量，但我还是会选择走10分钟到楼下的便利店里去买。这种出于经济方面的考虑（时间、精力、成本等）不仅在日常生活中很常见，在软件架构中也很重要。
这就是为什么ZMQ不会强制使用带有中间件的架构，但仍提供了像内置装置这样的中间件供编程人员*选用。
这一节我们会打破以往使用中间件进行可靠性设计的架构，转而使用点对点架构，即*者模式，来进行可靠的消息传输。我们的示例程序会是一个名称解析服务。ZMQ中的一个常见问题是：我们如何得知需要连接的端点？在代码中直接写入TCP/IP地址肯定是不合适的；使用配置文件会造成管理上的不便。试想一下，你要在上百台计算机中进行配置，只是为了让它们知道google.com的IP地址是74.125.230.82。
一个ZMQ的名称解析服务需要实现的功能有：
1、将逻辑名称解析为一个或多个端点地址，包括绑定端和连接端。实际使用时，名称服务会提供一组端点。
2、允许我们在不同的环境下，即开发环境和生产环境，进行解析。
3、该服务必须是可靠的，否则应用程序将无法连接到网络。
为管家模式提供名称解析服务会很有用，虽然将代理程序的端点对外暴露也很简单，但是如果用好名称解析服务，那它将成为唯一一个对外暴露的接口，将更便于管理。
我们需要处理的故障类型有：服务崩溃或重启、服务过载、网络因素等。为获取可靠性，我们必须建立一个服务群，当某个服务端崩溃后，客户端可以连接其他的服务端。实践中，两个服务端就已经足够了，但事实上服务端的数量可以是任意个。
在这个架构中，大量客户端和少量服务端进行通信，服务端将套接字绑定至单独的端口，这和管家模式中的代理有很大不同。对于客户端来说，它有这样几种选择：
1、客户端可以使用REQ套接字和懒惰海盗模式，但需要有一个机制防止客户端不断地请求已停止的服务端。
2、客户端可以使用DEALER套接字，向所有的服务端发送请求。很简单，但并不太妙；
3、客户端使用ROUTER套接字，连接特定的服务端。但客户端如何得知服务端的套接字标识呢？一种方式是让服务端主动连接客户端（很复杂），或者将服务端标识写入代码进行固化（很混乱）。
二、模型一：简单重试
让我们先尝试简单的方案，重写懒惰海盗模式，让其能够和多个服务端进行通信。启动服务端时用命令行参数指定端口。然后启动多个服务端。
flserver1: Freelance server, Model One in C
//
// *者模式 - 服务端 - 模型1
// 提供echo服务
//
#include "czmq.h"
int main (int argc, char *argv [])
{
if (argc < 2) {
printf ("I: syntax: %s <endpoint>\n", argv [0]);
exit (EXIT_SUCCESS);
}
zctx_t *ctx = zctx_new ();
void *server = zsocket_new (ctx, ZMQ_REP);
zsocket_bind (server, argv [1]);
printf ("I: echo服务端点： %s\n", argv [1]);
while (TRUE) {
zmsg_t *msg = zmsg_recv (server);
if (!msg)
break; // 中断
zmsg_send (&msg, server);
}
if (zctx_interrupted)
printf ("W: 中断\n");
zctx_destroy (&ctx);
return 0;
}
启动客户端，指定一个或多个端点：
flclient1: Freelance client, Model One in C
//
// *者模式 - 客户端 - 模型1
// 使用REQ套接字请求一个或多个服务端
//
#include "czmq.h"
#define REQUEST_TIMEOUT 1000
#define MAX_RETRIES 3 // 尝试次数
static zmsg_t *
s_try_request (zctx_t *ctx, char *endpoint, zmsg_t *request)
{
printf ("I: 在端点 %s 上尝试请求echo服务...\n", endpoint);
void *client = zsocket_new (ctx, ZMQ_REQ);
zsocket_connect (client, endpoint);
// 发送请求，并等待应答
zmsg_t *msg = zmsg_dup (request);
zmsg_send (&msg, client);
zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } };
zmq_poll (items, 1, REQUEST_TIMEOUT * ZMQ_POLL_MSEC);
zmsg_t *reply = NULL;
if (items [0].revents & ZMQ_POLLIN)
reply = zmsg_recv (client);
// 关闭套接字
zsocket_destroy (ctx, client);
return reply;
}
int main (int argc, char *argv [])
{
zctx_t *ctx = zctx_new ();
zmsg_t *request = zmsg_new ();
zmsg_addstr (request, "Hello world");
zmsg_t *reply = NULL;
int endpoints = argc - 1;
if (endpoints == 0)
printf ("I: syntax: %s <endpoint> ...\n", argv [0]);
else
if (endpoints == 1) {
// 若只有一个端点，则尝试N次
int retries;
for (retries = 0; retries < MAX_RETRIES; retries++) {
char *endpoint = argv [1];
reply = s_try_request (ctx, endpoint, request);
if (reply)
break; // 成功
printf ("W: 没有收到 %s 的应答, 准备重试...\n", endpoint);
}
}
else {
// 若有多个端点，则每个尝试一次
int endpoint_nbr;
for (endpoint_nbr = 0; endpoint_nbr < endpoints; endpoint_nbr++) {
char *endpoint = argv [endpoint_nbr + 1];
reply = s_try_request (ctx, endpoint, request);
if (reply)
break; // Successful
printf ("W: 没有收到 %s 的应答\n", endpoint);
}
}
if (reply)
printf ("服务运作正常\n");
zmsg_destroy (&request);
zmsg_destroy (&reply);
zctx_destroy (&ctx);
return 0;
}
可用如下命令运行：
flserver1 tcp://*:5555 &
flserver1 tcp://*:5556 &
flclient1 tcp://localhost:5555 tcp://localhost:5556
客户端的核心机制是懒惰海盗模式，即获得一次成功的应答后就结束。会有两种情况：
1、如果只有一个服务端，客户端会再尝试N次后停止，这和懒惰海盗模式的逻辑一致。
2、如果有多个服务端，客户端会每个尝试一次，收到应答后停止。
这种机制补充了海盗模式，使其能够克服只有一个服务端的情况。
但是，这种设计无法在现实程序中使用：当有很多客户端连接了服务端，而主服务端崩溃了，那所有客户端都需要在超时后才能继续执行。
三、模型二：批量发送
下面让我们使用DEALER套接字。我们的目标是能再最短的时间里收到一个应答，不能受主服务端崩溃的影响。可以采取以下措施：
1、连接所有的服务端。
2、当有请求时，一次性发送给所有的服务端。
3、等待第一个应答。
4、忽略其他应答。
这样设计客户端时，当发送请求后，所有的服务端都会收到这个请求，并返回应答。如果某个服务端断开连接了，ZMQ可能会将请求发给其他服务端，导致某些服务端会收到两次请求。
更麻烦的是客户端无法得知应答的数量，容易发生混乱。
我们可以为请求进行编号，忽略不匹配的应答。我们要对服务端进行改造，返回的消息中需要包含请求编号： flserver2: Freelance server, Model Two in C
//
// *者模式 - 服务端 - 模型2
// 返回带有请求编号的OK信息
//
#include "czmq.h"
int main (int argc, char *argv [])
{
if (argc < 2) {
printf ("I: syntax: %s <endpoint>\n", argv [0]);
exit (EXIT_SUCCESS);
}
zctx_t *ctx = zctx_new ();
void *server = zsocket_new (ctx, ZMQ_REP);
zsocket_bind (server, argv [1]);
printf ("I: 服务已就绪 %s\n", argv [1]);
while (TRUE) {
zmsg_t *request = zmsg_recv (server);
if (!request)
break; // 中断
// 判断请求内容是否正确
assert (zmsg_size (request) == 2);
zframe_t *address = zmsg_pop (request);
zmsg_destroy (&request);
zmsg_t *reply = zmsg_new ();
zmsg_add (reply, address);
zmsg_addstr (reply, "OK");
zmsg_send (&reply, server);
}
if (zctx_interrupted)
printf ("W: interrupted\n");
zctx_destroy (&ctx);
return 0;
}
客户端代码：
flclient2: Freelance client, Model Two in C
//
// *者模式 - 客户端 - 模型2
// 使用DEALER套接字发送批量消息
//
#include "czmq.h"
// 超时时间
#define GLOBAL_TIMEOUT 2500
// 将客户端API封装成一个类
#ifdef __cplusplus
extern "C" {
#endif
// 声明类结构
typedef struct _flclient_t flclient_t;
flclient_t *
flclient_new (void);
void
flclient_destroy (flclient_t **self_p);
void
flclient_connect (flclient_t *self, char *endpoint);
zmsg_t *
flclient_request (flclient_t *self, zmsg_t **request_p);
#ifdef __cplusplus
}
#endif
int main (int argc, char *argv [])
{
if (argc == 1) {
printf ("I: syntax: %s <endpoint> ...\n", argv [0]);
exit (EXIT_SUCCESS);
}
// 创建*者模式客户端
flclient_t *client = flclient_new ();
// 连接至各个端点
int argn;
for (argn = 1; argn < argc; argn++)
flclient_connect (client, argv [argn]);
// 发送一组请求，并记录时间
int requests = 10000;
uint64_t start = zclock_time ();
while (requests--) {
zmsg_t *request = zmsg_new ();
zmsg_addstr (request, "random name");
zmsg_t *reply = flclient_request (client, &request);
if (!reply) {
printf ("E: 名称解析服务不可用，正在退出\n");
break;
}
zmsg_destroy (&reply);
}
printf ("平均请求时间: %d 微秒\n",
(int) (zclock_time () - start) / 10);
flclient_destroy (&client);
return 0;
}
// --------------------------------------------------------------------
// 类结构
struct _flclient_t {
zctx_t *ctx; // 上下文
void *socket; // 用于和服务端通信的DEALER套接字
size_t servers; // 以连接的服务端数量
uint sequence; // 已发送的请求数
};
// --------------------------------------------------------------------
// Constructor
flclient_t *
flclient_new (void)
{
flclient_t
*self;
self = (flclient_t *) zmalloc (sizeof (flclient_t));
self->ctx = zctx_new ();
self->socket = zsocket_new (self->ctx, ZMQ_DEALER);
return self;
}
// --------------------------------------------------------------------
// 析构函数
void
flclient_destroy (flclient_t **self_p)
{
assert (self_p);
if (*self_p) {
flclient_t *self = *self_p;
zctx_destroy (&self->ctx);
free (self);
*self_p = NULL;
}
}
// --------------------------------------------------------------------
// 连接至新的服务端端点
void
flclient_connect (flclient_t *self, char *endpoint)
{
assert (self);
zsocket_connect (self->socket, endpoint);
self->servers++;
}
// --------------------------------------------------------------------
// 发送请求，接收应答
// 发送后销毁请求
zmsg_t *
flclient_request (flclient_t *self, zmsg_t **request_p)
{
assert (self);
assert (*request_p);
zmsg_t *request = *request_p;
// 向消息添加编号和空帧
char sequence_text [10];
sprintf (sequence_text, "%u", ++self->sequence);
zmsg_pushstr (request, sequence_text);
zmsg_pushstr (request, "");
// 向所有已连接的服务端发送请求
int server;
for (server = 0; server < self->servers; server++) {
zmsg_t *msg = zmsg_dup (request);
zmsg_send (&msg, self->socket);
}
// 接收来自任何服务端的应答
// 因为我们可能poll多次，所以每次都进行计算
zmsg_t *reply = NULL;
uint64_t endtime = zclock_time () + GLOBAL_TIMEOUT;
while (zclock_time () < endtime) {
zmq_pollitem_t items [] = { { self->socket, 0, ZMQ_POLLIN, 0 } };
zmq_poll (items, 1, (endtime - zclock_time ()) * ZMQ_POLL_MSEC);
if (items [0].revents & ZMQ_POLLIN) {
// 应答内容是 [empty][sequence][OK]
reply = zmsg_recv (self->socket);
assert (zmsg_size (reply) == 3);
free (zmsg_popstr (reply));
char *sequence = zmsg_popstr (reply);
int sequence_nbr = atoi (sequence);
free (sequence);
if (sequence_nbr == self->sequence)
break;
}
}
zmsg_destroy (request_p);
return reply;
}
几点说明：
1、客户端被封装成了一个API类，将复杂的代码都包装了起来。
2、户端会在几秒之后放弃寻找可用的服务端。
3、客户端需要创建一个合法的REP信封，所以需要添加一个空帧。
程序中，客户端发出了1万次名称解析请求（虽然是假的），并计算平均耗费时间。在我的测试机上，有一个服务端时，耗时60微妙；三个时80微妙。
该模型的优缺点是：
1、优点：简单，容易理解和编写。
2、优点：它工作迅速，有重试机制。
3、缺点：占用了额外的网络带宽。
4、缺点：我们不能为服务端设置优先级，如主服务、次服务等。
5、缺点：服务端不能同时处理多个请求。
四、模型三：Complex and Nasty
批量发送模型看起来不太真实，那就让我们来探索最后这个极度复杂的模型。很有可能在编写完之后我们又会转而使用批量发送，哈哈，这就是我的作风。
我们可以将客户端使用的套接字更换为ROUTER，让我们能够向特定的服务端发送请求，停止向已死亡的服务端发送请求，从而做得尽可能地智能。我们还可以将服务端的套接字更换为ROUTER，从而突破单线程的瓶颈。
但是，使用ROUTER-ROUTER套接字连接两个瞬时套接字是不可行的，节点只有在收到第一条消息时才会为对方生成套接字标识。唯一的方法是让其中一个节点使用持久化的套接字，比较好的方式是让客户端知道服务端的标识，即服务端作为持久化的套接字。
为了避免产生新的配置项，我们直接使用服务端的端点作为套接字标识。
回想一下ZMQ套接字标识是如何工作的。服务端的ROUTER套接字为自己设置一个标识（在绑定之前），当客户端连接时，通过一个握手的过程来交换双方的标识。客户端的ROUTER套接字会先发送一条空消息，服务端为客户端生成一个随机的UUID。然后，服务端会向客户端发送自己的标识。
这样一来，客户端就可以将消息发送给特定的服务端了。不过还有一个问题：我们不知道服务端会在什么时候完成这个握手的过程。如果服务端是在线的，那可能几毫秒就能完成。如果不在线，那可能需要很久很久。
这里有一个矛盾：我们需要知道服务端何时连接成功且能够开始工作。*者模式不像中间件模式，它的服务端必须要先发送请求后才能的应答。所以在服务端发送消息给客户端之前，客户端必须要先请求服务端，这看似是不可能的。
我有一个解决方法，那就是批量发送。这里发送的不是真正的请求，而是一个试探性的心跳（PING-PONG）。当收到应答时，就说明对方是在线的。
下面让我们制定一个协议，来定义*者模式是如何传递这种心跳的：10/FLP | ZeroMQ RFC
实现这个协议的服务端很方便，下面就是经过改造的echo服务：
flserver3: Freelance server, Model Three in C
//
// *者模式 - 服务端 - 模型3
// 使用ROUTER-ROUTER套接字进行通信；单线程。
//
#include "czmq.h"
int main (int argc, char *argv [])
{
int verbose = (argc > 1 && streq (argv [1], "-v"));
zctx_t *ctx = zctx_new ();
// 准备服务端套接字，其标识和端点名相同
char *bind_endpoint = "tcp://*:5555";
char *connect_endpoint = "tcp://localhost:5555";
void *server = zsocket_new (ctx, ZMQ_ROUTER);
zmq_setsockopt (server,
ZMQ_IDENTITY, connect_endpoint, strlen (connect_endpoint));
zsocket_bind (server, bind_endpoint);
printf ("I: 服务端已准备就绪 %s\n", bind_endpoint);
while (!zctx_interrupted) {
zmsg_t *request = zmsg_recv (server);
if (verbose && request)
zmsg_dump (request);
if (!request)
break; // 中断
// Frame 0: 客户端标识
// Frame 1: 心跳，或客户端控制信息帧
// Frame 2: 请求内容
zframe_t *address = zmsg_pop (request);
zframe_t *control = zmsg_pop (request);
zmsg_t *reply = zmsg_new ();
if (zframe_streq (control, "PONG"))
zmsg_addstr (reply, "PONG");
else {
zmsg_add (reply, control);
zmsg_addstr (reply, "OK");
}
zmsg_destroy (&request);
zmsg_push (reply, address);
if (verbose && reply)
zmsg_dump (reply);
zmsg_send (&reply, server);
}
if (zctx_interrupted)
printf ("W: 中断\n");
zctx_destroy (&ctx);
return 0;
}
但是，*者模式的客户端会变得大一写。为了清晰期间，我们将其拆分为两个类来实现。首先是在上层使用的程序：
flclient3: Freelance client, Model Three in C
//
// *者模式 - 客户端 - 模型3
// 使用flcliapi类来封装*者模式
//
// 直接编译，不建类库
#include "flcliapi.c"
int main (void)
{
// 创建*者模式实例
flcliapi_t *client = flcliapi_new ();
// 链接至服务器端点
flcliapi_connect (client, "tcp://localhost:5555");
flcliapi_connect (client, "tcp://localhost:5556");
flcliapi_connect (client, "tcp://localhost:5557");
// 发送随机请求，计算时间
int requests = 1000;
uint64_t start = zclock_time ();
while (requests--) {
zmsg_t *request = zmsg_new ();
zmsg_addstr (request, "random name");
zmsg_t *reply = flcliapi_request (client, &request);
if (!reply) {
printf ("E: 名称解析服务不可用，正在退出\n");
break;
}
zmsg_destroy (&reply);
}
printf ("平均执行时间： %d usec\n",
(int) (zclock_time () - start) / 10);
flcliapi_destroy (&client);
return 0;
}
下面是该模式复杂的实现过程：
flcliapi: Freelance client API in C
/* =====================================================================
flcliapi - Freelance Pattern agent class
Model 3: uses ROUTER socket to address specific services
---------------------------------------------------------------------
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 "flcliapi.h"
// 请求超时时间
#define GLOBAL_TIMEOUT 3000 // msecs
// 心跳间隔
#define PING_INTERVAL 2000 // msecs
// 判定服务死亡的时间
#define SERVER_TTL 6000 // msecs
// =====================================================================
// 同步部分，在应用程序层面运行
// ---------------------------------------------------------------------
// 类结构
struct _flcliapi_t {
zctx_t *ctx; // 上下文
void *pipe; // 用于和主线程通信的套接字
};
// 这是运行后台代理程序的线程
static void flcliapi_agent (void *args, zctx_t *ctx, void *pipe);
// ---------------------------------------------------------------------
// 构造函数
flcliapi_t *
flcliapi_new (void)
{
flcliapi_t
*self;
self = (flcliapi_t *) zmalloc (sizeof (flcliapi_t));
self->ctx = zctx_new ();
self->pipe = zthread_fork (self->ctx, flcliapi_agent, NULL);
return self;
}
// ---------------------------------------------------------------------
// 析构函数
void
flcliapi_destroy (flcliapi_t **self_p)
{
assert (self_p);
if (*self_p) {
flcliapi_t *self = *self_p;
zctx_destroy (&self->ctx);
free (self);
*self_p = NULL;
}
}
// ---------------------------------------------------------------------
// 连接至新服务器端点
// 消息内容：[CONNECT][endpoint]
void
flcliapi_connect (flcliapi_t *self, char *endpoint)
{
assert (self);
assert (endpoint);
zmsg_t *msg = zmsg_new ();
zmsg_addstr (msg, "CONNECT");
zmsg_addstr (msg, endpoint);
zmsg_send (&msg, self->pipe);
zclock_sleep (100); // 等待连接
}
// ---------------------------------------------------------------------
// 发送并销毁请求，接收应答
zmsg_t *
flcliapi_request (flcliapi_t *self, zmsg_t **request_p)
{
assert (self);
assert (*request_p);
zmsg_pushstr (*request_p, "REQUEST");
zmsg_send (request_p, self->pipe);
zmsg_t *reply = zmsg_recv (self->pipe);
if (reply) {
char *status = zmsg_popstr (reply);
if (streq (status, "FAILED"))
zmsg_destroy (&reply);
free (status);
}
return reply;
}
// =====================================================================
// 异步部分，在后台运行
// ---------------------------------------------------------------------
// 单个服务端信息
typedef struct {
char *endpoint; // 服务端端点/套接字标识
uint alive; // 是否在线
int64_t ping_at; // 下一次心跳时间
int64_t expires; // 过期时间
} server_t;
server_t *
server_new (char *endpoint)
{
server_t *self = (server_t *) zmalloc (sizeof (server_t));
self->endpoint = strdup (endpoint);
self->alive = 0;
self->ping_at = zclock_time () + PING_INTERVAL;
self->expires = zclock_time () + SERVER_TTL;
return self;
}
void
server_destroy (server_t **self_p)
{
assert (self_p);
if (*self_p) {
server_t *self = *self_p;
free (self->endpoint);
free (self);
*self_p = NULL;
}
}
int
server_ping (char *key, void *server, void *socket)
{
server_t *self = (server_t *) server;
if (zclock_time () >= self->ping_at) {
zmsg_t *ping = zmsg_new ();
zmsg_addstr (ping, self->endpoint);
zmsg_addstr (ping, "PING");
zmsg_send (&ping, socket);
self->ping_at = zclock_time () + PING_INTERVAL;
}
return 0;
}
int
server_tickless (char *key, void *server, void *arg)
{
server_t *self = (server_t *) server;
uint64_t *tickless = (uint64_t *) arg;
if (*tickless > self->ping_at)
*tickless = self->ping_at;
return 0;
}
// ---------------------------------------------------------------------
// 后台处理程序信息
typedef struct {
zctx_t *ctx; // 上下文
void *pipe; // 用于应用程序通信的套接字
void *router; // 用于服务端通信的套接字
zhash_t *servers; // 已连接的服务端
zlist_t *actives; // 在线的服务端
uint sequence; // 请求编号
zmsg_t *request; // 当前请求
zmsg_t *reply; // 当前应答
int64_t expires; // 请求过期时间
} agent_t;
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->router = zsocket_new (self->ctx, ZMQ_ROUTER);
self->servers = zhash_new ();
self->actives = zlist_new ();
return self;
}
void
agent_destroy (agent_t **self_p)
{
assert (self_p);
if (*self_p) {
agent_t *self = *self_p;
zhash_destroy (&self->servers);
zlist_destroy (&self->actives);
zmsg_destroy (&self->request);
zmsg_destroy (&self->reply);
free (self);
*self_p = NULL;
}
}
// 当服务端从列表中移除时，回调该函数。
static void
s_server_free (void *argument)
{
server_t *server = (server_t *) argument;
server_destroy (&server);
}
void
agent_control_message (agent_t *self)
{
zmsg_t *msg = zmsg_recv (self->pipe);
char *command = zmsg_popstr (msg);
if (streq (command, "CONNECT")) {
char *endpoint = zmsg_popstr (msg);
printf ("I: connecting to %s...\n", endpoint);
int rc = zmq_connect (self->router, endpoint);
assert (rc == 0);
server_t *server = server_new (endpoint);
zhash_insert (self->servers, endpoint, server);
zhash_freefn (self->servers, endpoint, s_server_free);
zlist_append (self->actives, server);
server->ping_at = zclock_time () + PING_INTERVAL;
server->expires = zclock_time () + SERVER_TTL;
free (endpoint);
}
else
if (streq (command, "REQUEST")) {
assert (!self->request); // 遵循请求-应答循环
// 将请求编号和空帧加入消息顶部
char sequence_text [10];
sprintf (sequence_text, "%u", ++self->sequence);
zmsg_pushstr (msg, sequence_text);
// 获取请求消息的所有权
self->request = msg;
msg = NULL;
// 设置请求过期时间
self->expires = zclock_time () + GLOBAL_TIMEOUT;
}
free (command);
zmsg_destroy (&msg);
}
void
agent_router_message (agent_t *self)
{
zmsg_t *reply = zmsg_recv (self->router);
// 第一帧是应答的服务端标识
char *endpoint = zmsg_popstr (reply);
server_t *server =
(server_t *) zhash_lookup (self->servers, endpoint);
assert (server);
free (endpoint);
if (!server->alive) {
zlist_append (self->actives, server);
server->alive = 1;
}
server->ping_at = zclock_time () + PING_INTERVAL;
server->expires = zclock_time () + SERVER_TTL;
// 第二帧是应答的编号
char *sequence = zmsg_popstr (reply);
if (atoi (sequence) == self->sequence) {
zmsg_pushstr (reply, "OK");
zmsg_send (&reply, self->pipe);
zmsg_destroy (&self->request);
}
else
zmsg_destroy (&reply);
}
// ---------------------------------------------------------------------
// 异步的后台代理会维护一个服务端池，处理请求和应答。
static void
flcliapi_agent (void *args, zctx_t *ctx, void *pipe)
{
agent_t *self = agent_new (ctx, pipe);
zmq_pollitem_t items [] = {
{ self->pipe, 0, ZMQ_POLLIN, 0 },
{ self->router, 0, ZMQ_POLLIN, 0 }
};
while (!zctx_interrupted) {
// 计算超时时间
uint64_t tickless = zclock_time () + 1000 * 3600;
if (self->request
&& tickless > self->expires)
tickless = self->expires;
zhash_foreach (self->servers, server_tickless, &tickless);
int rc = zmq_poll (items, 2,
(tickless - zclock_time ()) * ZMQ_POLL_MSEC);
if (rc == -1)
break; // 上下文对象被关闭
if (items [0].revents & ZMQ_POLLIN)
agent_control_message (self);
if (items [1].revents & ZMQ_POLLIN)
agent_router_message (self);
// 如果我们需要处理一项请求，将其发送给下一个可用的服务端
if (self->request) {
if (zclock_time () >= self->expires) {
// 请求超时
zstr_send (self->pipe, "FAILED");
zmsg_destroy (&self->request);
}
else {
// 寻找可用的服务端
while (zlist_size (self->actives)) {
server_t *server =
(server_t *) zlist_first (self->actives);
if (zclock_time () >= server->expires) {
zlist_pop (self->actives);
server->alive = 0;
}
else {
zmsg_t *request = zmsg_dup (self->request);
zmsg_pushstr (request, server->endpoint);
zmsg_send (&request, self->router);
break;
}
}
}
}
// 断开并删除已过期的服务端
// 发送心跳给空闲服务器
zhash_foreach (self->servers, server_ping, self->router);
}
agent_destroy (&self);
}