将 majordomo 代理与异步客户端一起使用答案

【问题标题】：Using the majordomo broker with asynchronous clients将 majordomo 代理与异步客户端一起使用
【发布时间】：2014-12-03 03:06:47
【问题描述】：

在阅读 zeromq 指南时，我遇到了客户端代码，它在一个循环中发送 100k 请求，然后在第二个循环中接收回复。

#include "../include/mdp.h"
#include <time.h>


int main (int argc, char *argv [])
{
    int verbose = (argc > 1 && streq (argv [1], "-v"));
    mdp_client_t *session = mdp_client_new ("tcp://localhost:5555", verbose);
    int count;
    for (count = 0; count < 100000; count++) {
        zmsg_t *request = zmsg_new ();
        zmsg_pushstr (request, "Hello world");
        mdp_client_send (session, "echo", &request);
    }
    printf("sent all\n");

    for (count = 0; count < 100000; count++) {
        zmsg_t *reply = mdp_client_recv (session,NULL,NULL);
        if (reply)
            zmsg_destroy (&reply);
        else
            break;              //  Interrupted by Ctrl-C
        printf("reply received:%d\n", count);
    }
    printf ("%d replies received\n", count);
    mdp_client_destroy (&session);
    return 0;
}

我添加了一个计数器来计算工作人员 (test_worker.c) 发送给代理的回复数量，并在 mdp_broker.c 中添加另一个计数器来计算代理发送给客户端的回复数量。这两个都计数到 100k，但客户端只收到大约 37k 回复。

如果客户端请求的数量设置为 40k 左右，则它会接收所有回复。当客户端发送超过 40k 的异步请求时，谁能告诉我为什么丢包？

我尝试将代理套接字的 HWM 设置为 100k，但问题仍然存在：

static broker_t *
s_broker_new (int verbose)
{
    broker_t *self = (broker_t *) zmalloc (sizeof (broker_t));
    int64_t hwm = 100000;
    //  Initialize broker state
    self->ctx = zctx_new ();
    self->socket = zsocket_new (self->ctx, ZMQ_ROUTER);
    zmq_setsockopt(self->socket, ZMQ_SNDHWM, &hwm, sizeof(hwm));

    zmq_setsockopt(self->socket, ZMQ_RCVHWM, &hwm, sizeof(hwm));
    self->verbose = verbose;
    self->services = zhash_new ();
    self->workers = zhash_new ();
    self->waiting = zlist_new ();
    self->heartbeat_at = zclock_time () + HEARTBEAT_INTERVAL;
    return self;
}

【问题讨论】：

标签： zeromq distributed-computing

【解决方案1】：

如果不设置 HWM 并使用默认 TCP 设置，仅 50k 条消息就会导致数据包丢失。

以下措施有助于减轻代理处的数据包丢失：

为 zeromq 套接字设置 HWM。
增加 TCP 发送/接收缓冲区大小。

这仅在一定程度上有所帮助。有两个客户端，每个发送 10 万条消息，代理能够很好地管理。但是当客户数量增加到三个时，他们就不再收到所有的回复了。

最后，帮助我确保不丢包的是通过以下方式更改客户端代码的设计：

客户端一次最多可以发送 N 条消息。客户端的 RCVHWM 和代理的 SNDHWM 应该足够高以容纳总共 N 条消息。
之后，对于客户端收到的每个回复，它都会发送两个请求。

【讨论】：

【解决方案2】：

您发送了 100k 条消息，然后开始接收它们。因此，这 100k 条消息应该存储在缓冲区中。当缓冲区耗尽且无法再存储消息时，您将达到 ZeroMQ 的高水位标记。 ZeroMQ 文档中指定了高水位线的行为。

在上述代码的情况下，代理可能会丢弃一些消息，因为a majordomo broker 使用了ROUTER socket。解决方案之一是将发送/接收循环拆分为单独的线程

【讨论】：

但是数据包丢失似乎发生在客户端。正如我在帖子中所说，经纪人向客户发送的回复数量总计达 100k。
是的，我认为这是受人尊敬的行为。代理收到了您计数的数据包。然后，它丢弃了我链接的文档中描述的数据包，因为它达到了高水位线。

【解决方案3】：

为什么输了？

在 ZeroMQ v2.1 中，ZMQ_HWM 的默认值是 INF（无穷大），这有助于上述测试有点意义，但代价是内存溢出崩溃的风险很大，因为缓冲区分配策略没有受到限制/控制以达到某些物理限制。

ZeroMQ v3.0+ 起，ZMQ_SNDHWM / ZMQ_RCVHWM 默认为 1000，之后可以设置。

您还可以阅读明确的警告，即

ØMQ 不保证套接字会接受与 ZMQ_SNDHWM 一样多的消息，实际限制可能会降低多达 60-70%取决于套接字上的消息流。

将发送/接收部分分成单独的线程有帮助吗？

没有。

快速修复？

是的，出于演示测试实验的目的，再次设置无限高水位线，但要小心避免在任何生产级软件中进行这种做法。

为什么要以这种方式测试 ZeroMQ 性能？

如上所述，最初的 demo-test 似乎在其 v2.1 实现中具有一定的意义。

从那时起，ZeroMQ 已经发展了很多。 step by step guide with code examples on ZeroMQ protocol overheads/performance case-study on large file transfers

...我们已经遇到了一个问题：如果我们向 ROUTER 套接字发送的数据过多，我们很容易将其溢出。简单但愚蠢的解决方案是在套接字上放置一个无限的高水位标记。这很愚蠢，因为我们现在无法防止耗尽服务器内存。然而，如果没有无限的 HWM，我们就有丢失大量大文件的风险。

试试这个：将 HWM 设置为 1,000（在 ZeroMQ v3.x 中这是默认值），然后将块大小减少到 100K，因此我们一次性发送 10K 块。运行测试，你会发现它永远不会结束。正如 zmq_socket() 手册页用欢快的粗暴说的那样，对于 ROUTER 套接字：“ZMQ_HWM 选项操作：Drop”。

我们必须预先控制服务器发送的数据量。发送超过网络可以处理的内容是没有意义的。让我们尝试一次发送一个块。在这个版本的协议中，客户端会明确地说“给我块 N”，然后服务器会从磁盘中获取特定的块并发送它。

据我所知，最好的部分是对“模型 3”流控制的结果性能的评论进展，人们可以从 ZeroMQ 指南中的精彩章节和真实评论中学到很多东西.

【讨论】：