zeroMQ初体验-29.可靠性-自由模式
好吧，本以为这可能是一个更靠谱的模式，谁知(其实是我一厢情愿了)。所谓自由模式，自然是--爱咋咋地呃。。
作为还算靠谱的官方，终是给出了一些经过验证的"自由模式"的模型。下面就来一一拆解吧。
简单重试和故障转移
好像有点眼熟，其实有用到"懒人模式"的简单重试，并且对服务器端做了重写，使其可适用于多服务器的情况。不过，这可不比之前的"主从模式"，而是由客户端连接多个服务器，一次性发出n个同样的请求，先得到谁的返回就用谁的应答。(真是有够简单的，难道是简陋？)
服务器端：
//// Freelance server - Model 2// Does some work, replies OK, with message sequencing//#include "czmq.h"int main (int argc, char *argv []){ if (argc < 2) { printf ("I: syntax: %s <endpoint>\n", argv ); exit (EXIT_SUCCESS); } zctx_t *ctx = zctx_new (); void *server = zsocket_new (ctx, ZMQ_REP); zsocket_bind (server, argv ); printf ("I: service is ready at %s\n", argv ); while (TRUE) { zmsg_t *request = zmsg_recv (server); if (!request) break; // Interrupted // Fail nastily if run against wrong client 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;
客户端：
//// Freelance client - Model 2// Uses DEALER socket to blast one or more services//#include "czmq.h"// If not a single service replies within this time, give up#define GLOBAL_TIMEOUT 2500// We design our client API as a class#ifdef __cplusplusextern "C" {#endif// Opaque class structuretypedef 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}#endifint main (int argc, char *argv []){ if (argc == 1) { printf ("I: syntax: %s <endpoint> …\n", argv ); exit (EXIT_SUCCESS); } // Create new freelance client object flclient_t *client = flclient_new (); // Connect to each endpoint int argn; for (argn = 1; argn < argc; argn++) flclient_connect (client, argv ); // Send a bunch of name resolution 'requests', measure time 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: name service not available, aborting\n"); break; } zmsg_destroy (&reply); } printf ("Average round trip cost: %d usec\n", (int) (zclock_time () - start) / 10); flclient_destroy (&client); return 0;}// --------------------------------------------------------------------// Structure of our classstruct _flclient_t { zctx_t *ctx; // Our context wrapper void *socket; // DEALER socket talking to servers size_t servers; // How many servers we have connected to uint sequence; // Number of requests ever sent};// --------------------------------------------------------------------// Constructorflclient_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;}// --------------------------------------------------------------------// Destructorvoidflclient_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; }}// --------------------------------------------------------------------// Connect to new server endpointvoidflclient_connect (flclient_t *self, char *endpoint){ assert (self); zsocket_connect (self->socket, endpoint); self->servers++;}// --------------------------------------------------------------------// Send request, get reply// Destroys request after sendingzmsg_t *flclient_request (flclient_t *self, zmsg_t **request_p){ assert (self); assert (*request_p); zmsg_t *request = *request_p; // Prefix request with sequence number and empty envelope char sequence_text ; sprintf (sequence_text, "%u", ++self->sequence); zmsg_pushstr (request, sequence_text); zmsg_pushstr (request, ""); // Blast the request to all connected servers int server; for (server = 0; server < self->servers; server++) { zmsg_t *msg = zmsg_dup (request); zmsg_send (&msg, self->socket); } // Wait for a matching reply to arrive from anywhere // Since we can poll several times, calculate each one 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 .revents & ZMQ_POLLIN) { // Reply is 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;}
客户端的代码做了些封装，使用到了api的引用结构，这也是作者所推崇的代码方式。
这种模型的优点：
1.简单！易于理解和使用
2.故障冗余不错(除非你的服务器全挂了)
缺点：
1.过多的冗余流量
2.服务器的使用没有优先级
3.所有的服务器同一时间内只能做一件事，还是同样的事
复杂到令人生厌
单词有一解释为“龌龊”。其实，可以理解为除了上面那种模型之外的模型。因为现实环境总是复杂的，那么模型相应也会越来越让人生厌，却又不得不如此。（果然龌龊）
通常来说，一个客户端知道一堆服务器模型虽然简单，可是业务运营未必，那么只让他知道一个路由地址呢。业务的代价就转移到技术上了(挺悲催的)。每多一层中间件，都会多一倍的复杂逻辑。
服务器端：
//// Freelance server - Model 3// Uses an ROUTER/ROUTER socket but just one thread//#include "czmq.h"int main (int argc, char *argv []){ int verbose = (argc > 1 && streq (argv , "-v")); zctx_t *ctx = zctx_new (); // Prepare server socket with predictable identity 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: service is ready at %s\n", bind_endpoint); while (!zctx_interrupted) { zmsg_t *request = zmsg_recv (server); if (verbose && request) zmsg_dump (request); if (!request) break; // Interrupted // Frame 0: identity of client // Frame 1: PING, or client control frame // Frame 2: request body 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: interrupted\n"); zctx_destroy (&ctx); return 0;}
客户端：
//// Freelance client - Model 3// Uses flcliapi class to encapsulate Freelance pattern//// Lets us build this source without creating a library#include "flcliapi.c"int main (void){ // Create new freelance client object flcliapi_t *client = flcliapi_new (); // Connect to several endpoints flcliapi_connect (client, "tcp://localhost:5555"); flcliapi_connect (client, "tcp://localhost:5556"); flcliapi_connect (client, "tcp://localhost:5557"); // Send a bunch of name resolution 'requests', measure time 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: name service not available, aborting\n"); break; } zmsg_destroy (&reply); } printf ("Average round trip cost: %d usec\n", (int) (zclock_time () - start) / 10);puts ("flclient 1"); flcliapi_destroy (&client);puts ("flclient 2"); return 0;}
客户端api:
/* ===================================================================== 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"// If no server replies within this time, abandon request#define GLOBAL_TIMEOUT 3000 // msecs// PING interval for servers we think are alive#define PING_INTERVAL 2000 // msecs// Server considered dead if silent for this long#define SERVER_TTL 6000 // msecs// =====================================================================// Synchronous part, works in our application thread// ---------------------------------------------------------------------// Structure of our classstruct _flcliapi_t { zctx_t *ctx; // Our context wrapper void *pipe; // Pipe through to flcliapi agent};// This is the thread that handles our real flcliapi classstatic void flcliapi_agent (void *args, zctx_t *ctx, void *pipe);// ---------------------------------------------------------------------// Constructorflcliapi_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;}// ---------------------------------------------------------------------// Destructorvoidflcliapi_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 to new server endpoint// Sends to the agentvoidflcliapi_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); // Allow connection to come up}// ---------------------------------------------------------------------// Send & destroy request, get replyzmsg_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;}// =====================================================================// Asynchronous part, works in the background// ---------------------------------------------------------------------// Simple class for one server we talk totypedef struct { char *endpoint; // Server identity/endpoint uint alive; // 1 if known to be alive int64_t ping_at; // Next ping at this time int64_t expires; // Expires at this time} 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;}voidserver_destroy (server_t **self_p){ assert (self_p); if (*self_p) { server_t *self = *self_p; free (self->endpoint); free (self); *self_p = NULL; }}intserver_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;}intserver_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;}// ---------------------------------------------------------------------// Simple class for one background agenttypedef struct { zctx_t *ctx; // Own context void *pipe; // Socket to talk back to application void *router; // Socket to talk to servers zhash_t *servers; // Servers we've connected to zlist_t *actives; // Servers we know are alive uint sequence; // Number of requests ever sent zmsg_t *request; // Current request if any zmsg_t *reply; // Current reply if any int64_t expires; // Timeout for request/reply} 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;}voidagent_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; }}// Callback when we remove server from agent 'servers' hash tablestatic voids_server_free (void *argument){ server_t *server = (server_t *) argument; server_destroy (&server);}voidagent_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); // Strict request-reply cycle // Prefix request with sequence number and empty envelope char sequence_text ; sprintf (sequence_text, "%u", ++self->sequence); zmsg_pushstr (msg, sequence_text); // Take ownership of request message self->request = msg; msg = NULL; // Request expires after global timeout self->expires = zclock_time () + GLOBAL_TIMEOUT; } free (command); zmsg_destroy (&msg);}voidagent_router_message (agent_t *self){ zmsg_t *reply = zmsg_recv (self->router); // Frame 0 is server that replied 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; // Frame 1 may be sequence number for reply 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);}// ---------------------------------------------------------------------// Asynchronous agent manages server pool and handles request/reply// dialog when the application asks for it.static voidflcliapi_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) { // Calculate tickless timer, up to 1 hour 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; // Context has been shut down if (items .revents & ZMQ_POLLIN) agent_control_message (self); if (items .revents & ZMQ_POLLIN) agent_router_message (self); // If we're processing a request, dispatch to next server if (self->request) { if (zclock_time () >= self->expires) { // Request expired, kill it zstr_send (self->pipe, "FAILED"); zmsg_destroy (&self->request); } else { // Find server to talk to, remove any expired ones 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; } } } } // Disconnect and delete any expired servers // Send heartbeats to idle servers if needed zhash_foreach (self->servers, server_ping, self->router); } agent_destroy (&self);}
呼呼，这个api有够复杂吧，它包含了：
异步机制
链路稳固机制
ping机制
计时器
终于，可靠性章节终于结束了。按官网的说法：比较靠谱，现实可用的就是上面那个令人生厌的模式了，当然“管家模式”也是可以一试的。
（未完待续）
页:
[1]