• ZMQ请求应答模式之无中间件的可靠性--自由者模式

    一、引言

            我们讲了那么多关于中间件的示例，好像有些违背“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. //
    2. //  自由者模式 - 服务端 - 模型1
    3. //  提供echo服务
    4. //
    5. #include "czmq.h"
    6.  
    7. int main (int argc, char *argv [])
    8. {
    9.     if (argc < 2) {
    10.         printf ("I: syntax: %s \n", argv [0]);
    11.         exit (EXIT_SUCCESS);
    12.     }
    13.     zctx_t *ctx = zctx_new ();
    14.     void *server = zsocket_new (ctx, ZMQ_REP);
    15.     zsocket_bind (server, argv [1]);
    16.  
    17.     printf ("I: echo服务端点： %s\n", argv [1]);
    18.     while (TRUE) {
    19.         zmsg_t *msg = zmsg_recv (server);
    20.         if (!msg)
    21.             break;          //  中断
    22.         zmsg_send (&msg, server);
    23.     }
    24.     if (zctx_interrupted)
    25.         printf ("W: 中断\n");
    26.  
    27.     zctx_destroy (&ctx);
    28.     return 0;
    29. }

            启动客户端，指定一个或多个端点：

            flclient1: Freelance client, Model One in C

    1. //
    2. //  自由者模式 - 客户端 - 模型1
    3. //  使用REQ套接字请求一个或多个服务端
    4. //
    5. #include "czmq.h"
    6.  
    7. #define REQUEST_TIMEOUT     1000
    8. #define MAX_RETRIES         3       //  尝试次数
    9.  
    10.  
    11. static zmsg_t *
    12. s_try_request (zctx_t *ctx, char *endpoint, zmsg_t *request)
    13. {
    14.     printf ("I: 在端点 %s 上尝试请求echo服务...\n", endpoint);
    15.     void *client = zsocket_new (ctx, ZMQ_REQ);
    16.     zsocket_connect (client, endpoint);
    17.  
    18.     //  发送请求，并等待应答
    19.     zmsg_t *msg = zmsg_dup (request);
    20.     zmsg_send (&msg, client);
    21.     zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } };
    22.     zmq_poll (items, 1, REQUEST_TIMEOUT * ZMQ_POLL_MSEC);
    23.     zmsg_t *reply = NULL;
    24.     if (items [0].revents & ZMQ_POLLIN)
    25.         reply = zmsg_recv (client);
    26.  
    27.     //  关闭套接字
    28.     zsocket_destroy (ctx, client);
    29.     return reply;
    30. }
    31.  
    32.  
    33. int main (int argc, char *argv [])
    34. {
    35.     zctx_t *ctx = zctx_new ();
    36.     zmsg_t *request = zmsg_new ();
    37.     zmsg_addstr (request, "Hello world");
    38.     zmsg_t *reply = NULL;
    39.  
    40.     int endpoints = argc - 1;
    41.     if (endpoints == 0)
    42.         printf ("I: syntax: %s  ...\n", argv [0]);
    43.     else
    44.     if (endpoints == 1) {
    45.         //  若只有一个端点，则尝试N次
    46.         int retries;
    47.         for (retries = 0; retries < MAX_RETRIES; retries++) {
    48.             char *endpoint = argv [1];
    49.             reply = s_try_request (ctx, endpoint, request);
    50.             if (reply)
    51.                 break;          //  成功
    52.             printf ("W: 没有收到 %s 的应答, 准备重试...\n", endpoint);
    53.         }
    54.     }
    55.     else {
    56.         //  若有多个端点，则每个尝试一次
    57.         int endpoint_nbr;
    58.         for (endpoint_nbr = 0; endpoint_nbr < endpoints; endpoint_nbr++) {
    59.             char *endpoint = argv [endpoint_nbr + 1];
    60.             reply = s_try_request (ctx, endpoint, request);
    61.             if (reply)
    62.                 break;          //  Successful
    63.             printf ("W: 没有收到 %s 的应答\n", endpoint);
    64.         }
    65.     }
    66.     if (reply)
    67.         printf ("服务运作正常\n");
    68.  
    69.     zmsg_destroy (&request);
    70.     zmsg_destroy (&reply);
    71.     zctx_destroy (&ctx);
    72.     return 0;
    73. }

            可用如下命令运行：

    1. flserver1 tcp://*:5555 &
    2. flserver1 tcp://*:5556 &
    3. flclient1 tcp://localhost:5555 tcp://localhost:5556

            客户端的核心机制是懒惰海盗模式，即获得一次成功的应答后就结束。会有两种情况：

                    1、如果只有一个服务端，客户端会再尝试N次后停止，这和懒惰海盗模式的逻辑一致。

                    2、如果有多个服务端，客户端会每个尝试一次，收到应答后停止。

            这种机制补充了海盗模式，使其能够克服只有一个服务端的情况。

            但是，这种设计无法在现实程序中使用：当有很多客户端连接了服务端，而主服务端崩溃了，那所有客户端都需要在超时后才能继续执行。

    三、模型二：批量发送

            下面让我们使用DEALER套接字。我们的目标是能再最短的时间里收到一个应答，不能受主服务端崩溃的影响。可以采取以下措施：

                    1、连接所有的服务端。

                    2、当有请求时，一次性发送给所有的服务端。

                    3、等待第一个应答。

                    4、忽略其他应答。

            这样设计客户端时，当发送请求后，所有的服务端都会收到这个请求，并返回应答。如果某个服务端断开连接了，ZMQ可能会将请求发给其他服务端，导致某些服务端会收到两次请求。

            更麻烦的是客户端无法得知应答的数量，容易发生混乱。

            我们可以为请求进行编号，忽略不匹配的应答。我们要对服务端进行改造，返回的消息中需要包含请求编号：
            flserver2: Freelance server, Model Two in C

    1. //
    2. //  自由者模式 - 服务端 - 模型2
    3. //  返回带有请求编号的OK信息
    4. //
    5. #include "czmq.h"
    6.  
    7. int main (int argc, char *argv [])
    8. {
    9.     if (argc < 2) {
    10.         printf ("I: syntax: %s \n", argv [0]);
    11.         exit (EXIT_SUCCESS);
    12.     }
    13.     zctx_t *ctx = zctx_new ();
    14.     void *server = zsocket_new (ctx, ZMQ_REP);
    15.     zsocket_bind (server, argv [1]);
    16.  
    17.     printf ("I: 服务已就绪 %s\n", argv [1]);
    18.     while (TRUE) {
    19.         zmsg_t *request = zmsg_recv (server);
    20.         if (!request)
    21.             break;          //  中断
    22.         //  判断请求内容是否正确
    23.         assert (zmsg_size (request) == 2);
    24.  
    25.         zframe_t *address = zmsg_pop (request);
    26.         zmsg_destroy (&request);
    27.  
    28.         zmsg_t *reply = zmsg_new ();
    29.         zmsg_add (reply, address);
    30.         zmsg_addstr (reply, "OK");
    31.         zmsg_send (&reply, server);
    32.     }
    33.     if (zctx_interrupted)
    34.         printf ("W: interrupted\n");
    35.  
    36.     zctx_destroy (&ctx);
    37.     return 0;
    38. }

            客户端代码：

            flclient2: Freelance client, Model Two in C

    1. //
    2. //  自由者模式 - 客户端 - 模型2
    3. //  使用DEALER套接字发送批量消息
    4. //
    5. #include "czmq.h"
    6.  
    7. //  超时时间
    8. #define GLOBAL_TIMEOUT 2500
    9.  
    10. //  将客户端API封装成一个类
    11.  
    12. #ifdef __cplusplus
    13. extern "C" {
    14. #endif
    15.  
    16. //  声明类结构
    17. typedef struct _flclient_t flclient_t;
    18.  
    19. flclient_t *
    20.     flclient_new (void);
    21. void
    22.     flclient_destroy (flclient_t **self_p);
    23. void
    24.     flclient_connect (flclient_t *self, char *endpoint);
    25. zmsg_t *
    26.     flclient_request (flclient_t *self, zmsg_t **request_p);
    27.  
    28. #ifdef __cplusplus
    29. }
    30. #endif
    31.  
    32.  
    33. int main (int argc, char *argv [])
    34. {
    35.     if (argc == 1) {
    36.         printf ("I: syntax: %s  ...\n", argv [0]);
    37.         exit (EXIT_SUCCESS);
    38.     }
    39.     //  创建自由者模式客户端
    40.     flclient_t *client = flclient_new ();
    41.  
    42.     //  连接至各个端点
    43.     int argn;
    44.     for (argn = 1; argn < argc; argn++)
    45.         flclient_connect (client, argv [argn]);
    46.  
    47.     //  发送一组请求，并记录时间
    48.     int requests = 10000;
    49.     uint64_t start = zclock_time ();
    50.     while (requests--) {
    51.         zmsg_t *request = zmsg_new ();
    52.         zmsg_addstr (request, "random name");
    53.         zmsg_t *reply = flclient_request (client, &request);
    54.         if (!reply) {
    55.             printf ("E: 名称解析服务不可用，正在退出\n");
    56.             break;
    57.         }
    58.         zmsg_destroy (&reply);
    59.     }
    60.     printf ("平均请求时间: %d 微秒\n",
    61.         (int) (zclock_time () - start) / 10);
    62.  
    63.     flclient_destroy (&client);
    64.     return 0;
    65. }
    66.  
    67.  
    68.  
    69. //  --------------------------------------------------------------------
    70. //  类结构
    71.  
    72. struct _flclient_t {
    73.     zctx_t *ctx;        //  上下文
    74.     void *socket;       //  用于和服务端通信的DEALER套接字
    75.     size_t servers;     //  以连接的服务端数量
    76.     uint sequence;      //  已发送的请求数
    77. };
    78.  
    79.  
    80. //  --------------------------------------------------------------------
    81. //  Constructor
    82.  
    83. flclient_t *
    84. flclient_new (void)
    85. {
    86.     flclient_t
    87.         *self;
    88.  
    89.     self = (flclient_t *) zmalloc (sizeof (flclient_t));
    90.     self->ctx = zctx_new ();
    91.     self->socket = zsocket_new (self->ctx, ZMQ_DEALER);
    92.     return self;
    93. }
    94.  
    95. //  --------------------------------------------------------------------
    96. //  析构函数
    97.  
    98. void
    99. flclient_destroy (flclient_t **self_p)
    100. {
    101.     assert (self_p);
    102.     if (*self_p) {
    103.         flclient_t *self = *self_p;
    104.         zctx_destroy (&self->ctx);
    105.         free (self);
    106.         *self_p = NULL;
    107.     }
    108. }
    109.  
    110. //  --------------------------------------------------------------------
    111. //  连接至新的服务端端点
    112.  
    113. void
    114. flclient_connect (flclient_t *self, char *endpoint)
    115. {
    116.     assert (self);
    117.     zsocket_connect (self->socket, endpoint);
    118.     self->servers++;
    119. }
    120.  
    121. //  --------------------------------------------------------------------
    122. //  发送请求，接收应答
    123. //  发送后销毁请求
    124.  
    125. zmsg_t *
    126. flclient_request (flclient_t *self, zmsg_t **request_p)
    127. {
    128.     assert (self);
    129.     assert (*request_p);
    130.     zmsg_t *request = *request_p;
    131.  
    132.     //  向消息添加编号和空帧
    133.     char sequence_text [10];
    134.     sprintf (sequence_text, "%u", ++self->sequence);
    135.     zmsg_pushstr (request, sequence_text);
    136.     zmsg_pushstr (request, "");
    137.  
    138.     //  向所有已连接的服务端发送请求
    139.     int server;
    140.     for (server = 0; server < self->servers; server++) {
    141.         zmsg_t *msg = zmsg_dup (request);
    142.         zmsg_send (&msg, self->socket);
    143.     }
    144.     //  接收来自任何服务端的应答
    145.     //  因为我们可能poll多次，所以每次都进行计算
    146.     zmsg_t *reply = NULL;
    147.     uint64_t endtime = zclock_time () + GLOBAL_TIMEOUT;
    148.     while (zclock_time () < endtime) {
    149.         zmq_pollitem_t items [] = { { self->socket, 0, ZMQ_POLLIN, 0 } };
    150.         zmq_poll (items, 1, (endtime - zclock_time ()) * ZMQ_POLL_MSEC);
    151.         if (items [0].revents & ZMQ_POLLIN) {
    152.             //  应答内容是 [empty][sequence][OK]
    153.             reply = zmsg_recv (self->socket);
    154.             assert (zmsg_size (reply) == 3);
    155.             free (zmsg_popstr (reply));
    156.             char *sequence = zmsg_popstr (reply);
    157.             int sequence_nbr = atoi (sequence);
    158.             free (sequence);
    159.             if (sequence_nbr == self->sequence)
    160.                 break;
    161.         }
    162.     }
    163.     zmsg_destroy (request_p);
    164.     return reply;
    165. }

            几点说明：

                    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

    1. //
    2. //  自由者模式 - 服务端 - 模型3
    3. //  使用ROUTER-ROUTER套接字进行通信；单线程。
    4. //
    5. #include "czmq.h"
    6.  
    7. int main (int argc, char *argv [])
    8. {
    9.     int verbose = (argc > 1 && streq (argv [1], "-v"));
    10.  
    11.     zctx_t *ctx = zctx_new ();
    12.  
    13.     //  准备服务端套接字，其标识和端点名相同
    14.     char *bind_endpoint = "tcp://*:5555";
    15.     char *connect_endpoint = "tcp://localhost:5555";
    16.     void *server = zsocket_new (ctx, ZMQ_ROUTER);
    17.     zmq_setsockopt (server,
    18.         ZMQ_IDENTITY, connect_endpoint, strlen (connect_endpoint));
    19.     zsocket_bind (server, bind_endpoint);
    20.     printf ("I: 服务端已准备就绪 %s\n", bind_endpoint);
    21.  
    22.     while (!zctx_interrupted) {
    23.         zmsg_t *request = zmsg_recv (server);
    24.         if (verbose && request)
    25.             zmsg_dump (request);
    26.         if (!request)
    27.             break;          //  中断
    28.  
    29.         //  Frame 0: 客户端标识
    30.         //  Frame 1: 心跳，或客户端控制信息帧
    31.         //  Frame 2: 请求内容
    32.         zframe_t *address = zmsg_pop (request);
    33.         zframe_t *control = zmsg_pop (request);
    34.         zmsg_t *reply = zmsg_new ();
    35.         if (zframe_streq (control, "PONG"))
    36.             zmsg_addstr (reply, "PONG");
    37.         else {
    38.             zmsg_add (reply, control);
    39.             zmsg_addstr (reply, "OK");
    40.         }
    41.         zmsg_destroy (&request);
    42.         zmsg_push (reply, address);
    43.         if (verbose && reply)
    44.             zmsg_dump (reply);
    45.         zmsg_send (&reply, server);
    46.     }
    47.     if (zctx_interrupted)
    48.         printf ("W: 中断\n");
    49.  
    50.     zctx_destroy (&ctx);
    51.     return 0;
    52. }

            但是，自由者模式的客户端会变得大一写。为了清晰期间，我们将其拆分为两个类来实现。首先是在上层使用的程序：

            flclient3: Freelance client, Model Three in C

    1. //
    2. //  自由者模式 - 客户端 - 模型3
    3. //  使用flcliapi类来封装自由者模式
    4. //
    5. //  直接编译，不建类库
    6. #include "flcliapi.c"
    7.  
    8. int main (void)
    9. {
    10.     //  创建自由者模式实例
    11.     flcliapi_t *client = flcliapi_new ();
    12.  
    13.     //  链接至服务器端点
    14.     flcliapi_connect (client, "tcp://localhost:5555");
    15.     flcliapi_connect (client, "tcp://localhost:5556");
    16.     flcliapi_connect (client, "tcp://localhost:5557");
    17.  
    18.     //  发送随机请求，计算时间
    19.     int requests = 1000;
    20.     uint64_t start = zclock_time ();
    21.     while (requests--) {
    22.         zmsg_t *request = zmsg_new ();
    23.         zmsg_addstr (request, "random name");
    24.         zmsg_t *reply = flcliapi_request (client, &request);
    25.         if (!reply) {
    26.             printf ("E: 名称解析服务不可用，正在退出\n");
    27.             break;
    28.         }
    29.         zmsg_destroy (&reply);
    30.     }
    31.     printf ("平均执行时间： %d usec\n",
    32.         (int) (zclock_time () - start) / 10);
    33.  
    34.     flcliapi_destroy (&client);
    35.     return 0;
    36. }

            下面是该模式复杂的实现过程：

            flcliapi: Freelance client API in C

    1. /*  =====================================================================
    2.     flcliapi - Freelance Pattern agent class
    3.     Model 3: uses ROUTER socket to address specific services
    4.  
    5.     ---------------------------------------------------------------------
    6.     Copyright (c) 1991-2011 iMatix Corporation 
    7.     Copyright other contributors as noted in the AUTHORS file.
    8.  
    9.     This file is part of the ZeroMQ Guide: http://zguide.zeromq.org
    10.  
    11.     This is free software; you can redistribute it and/or modify it under
    12.     the terms of the GNU Lesser General Public License as published by
    13.     the Free Software Foundation; either version 3 of the License, or (at
    14.     your option) any later version.
    15.  
    16.     This software is distributed in the hope that it will be useful, but
    17.     WITHOUT ANY WARRANTY; without even the implied warranty of
    18.     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
    19.     Lesser General Public License for more details.
    20.  
    21.     You should have received a copy of the GNU Lesser General Public
    22.     License along with this program. If not, see
    23.     .
    24.     =====================================================================
    25. */
    26.  
    27. #include "flcliapi.h"
    28.  
    29. //  请求超时时间
    30. #define GLOBAL_TIMEOUT  3000    //  msecs
    31. //  心跳间隔
    32. #define PING_INTERVAL   2000    //  msecs
    33. //  判定服务死亡的时间
    34. #define SERVER_TTL      6000    //  msecs
    35.  
    36.  
    37. //  =====================================================================
    38. //  同步部分，在应用程序层面运行
    39.  
    40. //  ---------------------------------------------------------------------
    41. //  类结构
    42.  
    43. struct _flcliapi_t {
    44.     zctx_t *ctx;        //  上下文
    45.     void *pipe;         //  用于和主线程通信的套接字
    46. };
    47.  
    48. //  这是运行后台代理程序的线程
    49. static void flcliapi_agent (void *args, zctx_t *ctx, void *pipe);
    50.  
    51.  
    52. //  ---------------------------------------------------------------------
    53. //  构造函数
    54.  
    55. flcliapi_t *
    56. flcliapi_new (void)
    57. {
    58.     flcliapi_t
    59.         *self;
    60.  
    61.     self = (flcliapi_t *) zmalloc (sizeof (flcliapi_t));
    62.     self->ctx = zctx_new ();
    63.     self->pipe = zthread_fork (self->ctx, flcliapi_agent, NULL);
    64.     return self;
    65. }
    66.  
    67. //  ---------------------------------------------------------------------
    68. //  析构函数
    69.  
    70. void
    71. flcliapi_destroy (flcliapi_t **self_p)
    72. {
    73.     assert (self_p);
    74.     if (*self_p) {
    75.         flcliapi_t *self = *self_p;
    76.         zctx_destroy (&self->ctx);
    77.         free (self);
    78.         *self_p = NULL;
    79.     }
    80. }
    81.  
    82. //  ---------------------------------------------------------------------
    83. //  连接至新服务器端点
    84. //  消息内容：[CONNECT][endpoint]
    85.  
    86. void
    87. flcliapi_connect (flcliapi_t *self, char *endpoint)
    88. {
    89.     assert (self);
    90.     assert (endpoint);
    91.     zmsg_t *msg = zmsg_new ();
    92.     zmsg_addstr (msg, "CONNECT");
    93.     zmsg_addstr (msg, endpoint);
    94.     zmsg_send (&msg, self->pipe);
    95.     zclock_sleep (100);      //  等待连接
    96. }
    97.  
    98. //  ---------------------------------------------------------------------
    99. //  发送并销毁请求，接收应答
    100.  
    101. zmsg_t *
    102. flcliapi_request (flcliapi_t *self, zmsg_t **request_p)
    103. {
    104.     assert (self);
    105.     assert (*request_p);
    106.  
    107.     zmsg_pushstr (*request_p, "REQUEST");
    108.     zmsg_send (request_p, self->pipe);
    109.     zmsg_t *reply = zmsg_recv (self->pipe);
    110.     if (reply) {
    111.         char *status = zmsg_popstr (reply);
    112.         if (streq (status, "FAILED"))
    113.             zmsg_destroy (&reply);
    114.         free (status);
    115.     }
    116.     return reply;
    117. }
    118.  
    119.  
    120. //  =====================================================================
    121. //  异步部分，在后台运行
    122.  
    123. //  ---------------------------------------------------------------------
    124. //  单个服务端信息
    125.  
    126. typedef struct {
    127.     char *endpoint;             //  服务端端点/套接字标识
    128.     uint alive;                 //  是否在线
    129.     int64_t ping_at;            //  下一次心跳时间
    130.     int64_t expires;            //  过期时间
    131. } server_t;
    132.  
    133. server_t *
    134. server_new (char *endpoint)
    135. {
    136.     server_t *self = (server_t *) zmalloc (sizeof (server_t));
    137.     self->endpoint = strdup (endpoint);
    138.     self->alive = 0;
    139.     self->ping_at = zclock_time () + PING_INTERVAL;
    140.     self->expires = zclock_time () + SERVER_TTL;
    141.     return self;
    142. }
    143.  
    144. void
    145. server_destroy (server_t **self_p)
    146. {
    147.     assert (self_p);
    148.     if (*self_p) {
    149.         server_t *self = *self_p;
    150.         free (self->endpoint);
    151.         free (self);
    152.         *self_p = NULL;
    153.     }
    154. }
    155.  
    156. int
    157. server_ping (char *key, void *server, void *socket)
    158. {
    159.     server_t *self = (server_t *) server;
    160.     if (zclock_time () >= self->ping_at) {
    161.         zmsg_t *ping = zmsg_new ();
    162.         zmsg_addstr (ping, self->endpoint);
    163.         zmsg_addstr (ping, "PING");
    164.         zmsg_send (&ping, socket);
    165.         self->ping_at = zclock_time () + PING_INTERVAL;
    166.     }
    167.     return 0;
    168. }
    169.  
    170. int
    171. server_tickless (char *key, void *server, void *arg)
    172. {
    173.     server_t *self = (server_t *) server;
    174.     uint64_t *tickless = (uint64_t *) arg;
    175.     if (*tickless > self->ping_at)
    176.         *tickless = self->ping_at;
    177.     return 0;
    178. }
    179.  
    180.  
    181. //  ---------------------------------------------------------------------
    182. //  后台处理程序信息
    183.  
    184. typedef struct {
    185.     zctx_t *ctx;                //  上下文
    186.     void *pipe;                 //  用于应用程序通信的套接字
    187.     void *router;               //  用于服务端通信的套接字
    188.     zhash_t *servers;           //  已连接的服务端
    189.     zlist_t *actives;           //  在线的服务端
    190.     uint sequence;              //  请求编号
    191.     zmsg_t *request;            //  当前请求
    192.     zmsg_t *reply;              //  当前应答
    193.     int64_t expires;            //  请求过期时间
    194. } agent_t;
    195.  
    196. agent_t *
    197. agent_new (zctx_t *ctx, void *pipe)
    198. {
    199.     agent_t *self = (agent_t *) zmalloc (sizeof (agent_t));
    200.     self->ctx = ctx;
    201.     self->pipe = pipe;
    202.     self->router = zsocket_new (self->ctx, ZMQ_ROUTER);
    203.     self->servers = zhash_new ();
    204.     self->actives = zlist_new ();
    205.     return self;
    206. }
    207.  
    208. void
    209. agent_destroy (agent_t **self_p)
    210. {
    211.     assert (self_p);
    212.     if (*self_p) {
    213.         agent_t *self = *self_p;
    214.         zhash_destroy (&self->servers);
    215.         zlist_destroy (&self->actives);
    216.         zmsg_destroy (&self->request);
    217.         zmsg_destroy (&self->reply);
    218.         free (self);
    219.         *self_p = NULL;
    220.     }
    221. }
    222.  
    223. //  当服务端从列表中移除时，回调该函数。
    224.  
    225. static void
    226. s_server_free (void *argument)
    227. {
    228.     server_t *server = (server_t *) argument;
    229.     server_destroy (&server);
    230. }
    231.  
    232. void
    233. agent_control_message (agent_t *self)
    234. {
    235.     zmsg_t *msg = zmsg_recv (self->pipe);
    236.     char *command = zmsg_popstr (msg);
    237.  
    238.     if (streq (command, "CONNECT")) {
    239.         char *endpoint = zmsg_popstr (msg);
    240.         printf ("I: connecting to %s...\n", endpoint);
    241.         int rc = zmq_connect (self->router, endpoint);
    242.         assert (rc == 0);
    243.         server_t *server = server_new (endpoint);
    244.         zhash_insert (self->servers, endpoint, server);
    245.         zhash_freefn (self->servers, endpoint, s_server_free);
    246.         zlist_append (self->actives, server);
    247.         server->ping_at = zclock_time () + PING_INTERVAL;
    248.         server->expires = zclock_time () + SERVER_TTL;
    249.         free (endpoint);
    250.     }
    251.     else
    252.     if (streq (command, "REQUEST")) {
    253.         assert (!self->request);    //  遵循请求-应答循环
    254.         //  将请求编号和空帧加入消息顶部
    255.         char sequence_text [10];
    256.         sprintf (sequence_text, "%u", ++self->sequence);
    257.         zmsg_pushstr (msg, sequence_text);
    258.         //  获取请求消息的所有权
    259.         self->request = msg;
    260.         msg = NULL;
    261.         //  设置请求过期时间
    262.         self->expires = zclock_time () + GLOBAL_TIMEOUT;
    263.     }
    264.     free (command);
    265.     zmsg_destroy (&msg);
    266. }
    267.  
    268. void
    269. agent_router_message (agent_t *self)
    270. {
    271.     zmsg_t *reply = zmsg_recv (self->router);
    272.  
    273.     //  第一帧是应答的服务端标识
    274.     char *endpoint = zmsg_popstr (reply);
    275.     server_t *server =
    276.         (server_t *) zhash_lookup (self->servers, endpoint);
    277.     assert (server);
    278.     free (endpoint);
    279.     if (!server->alive) {
    280.         zlist_append (self->actives, server);
    281.         server->alive = 1;
    282.     }
    283.     server->ping_at = zclock_time () + PING_INTERVAL;
    284.     server->expires = zclock_time () + SERVER_TTL;
    285.  
    286.     //  第二帧是应答的编号
    287.     char *sequence = zmsg_popstr (reply);
    288.     if (atoi (sequence) == self->sequence) {
    289.         zmsg_pushstr (reply, "OK");
    290.         zmsg_send (&reply, self->pipe);
    291.         zmsg_destroy (&self->request);
    292.     }
    293.     else
    294.         zmsg_destroy (&reply);
    295. }
    296.  
    297.  
    298. //  ---------------------------------------------------------------------
    299. //  异步的后台代理会维护一个服务端池，处理请求和应答。
    300.  
    301. static void
    302. flcliapi_agent (void *args, zctx_t *ctx, void *pipe)
    303. {
    304.     agent_t *self = agent_new (ctx, pipe);
    305.  
    306.     zmq_pollitem_t items [] = {
    307.         { self->pipe, 0, ZMQ_POLLIN, 0 },
    308.         { self->router, 0, ZMQ_POLLIN, 0 }
    309.     };
    310.     while (!zctx_interrupted) {
    311.         //  计算超时时间
    312.         uint64_t tickless = zclock_time () + 1000 * 3600;
    313.         if (self->request
    314.         &&  tickless > self->expires)
    315.             tickless = self->expires;
    316.         zhash_foreach (self->servers, server_tickless, &tickless);
    317.  
    318.         int rc = zmq_poll (items, 2,
    319.             (tickless - zclock_time ()) * ZMQ_POLL_MSEC);
    320.         if (rc == -1)
    321.             break;              //  上下文对象被关闭
    322.  
    323.         if (items [0].revents & ZMQ_POLLIN)
    324.             agent_control_message (self);
    325.  
    326.         if (items [1].revents & ZMQ_POLLIN)
    327.             agent_router_message (self);
    328.  
    329.         //  如果我们需要处理一项请求，将其发送给下一个可用的服务端
    330.         if (self->request) {
    331.             if (zclock_time () >= self->expires) {
    332.                 //  请求超时
    333.                 zstr_send (self->pipe, "FAILED");
    334.                 zmsg_destroy (&self->request);
    335.             }
    336.             else {
    337.                 //  寻找可用的服务端
    338.                 while (zlist_size (self->actives)) {
    339.                     server_t *server =
    340.                         (server_t *) zlist_first (self->actives);
    341.                     if (zclock_time () >= server->expires) {
    342.                         zlist_pop (self->actives);
    343.                         server->alive = 0;
    344.                     }
    345.                     else {
    346.                         zmsg_t *request = zmsg_dup (self->request);
    347.                         zmsg_pushstr (request, server->endpoint);
    348.                         zmsg_send (&request, self->router);
    349.                         break;
    350.                     }
    351.                 }
    352.             }
    353.         }
    354.         //  断开并删除已过期的服务端
    355.         //  发送心跳给空闲服务器
    356.         zhash_foreach (self->servers, server_ping, self->router);
    357.     }
    358.     agent_destroy (&self);
    359. }

  • 相关阅读:
    路由vue-route的使用
    【微信开发第四章】SpringBoot实现微信H5支付
    git基本命令
    看万山红遍
    [附源码]计算机毕业设计springboot第三方游戏零售平台
    Listener(监听器)-ServletContextListener
    接口自动化之测试数据动态生成并替换
    JAVA代码 企业人力资源管理系统(详细带截图) 毕业设计
    Unity UI Toolkit学习笔记-EditorWindow
    想当领导，有哪些能力是需要培训的？
  • 原文地址：https://blog.csdn.net/code_lyb/article/details/128119372