zeromq - 将 Majordomo 代理与异步客户端一起使用

Question

在阅读 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;
}

Answer 1

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

以下有助于减轻代理的数据包丢失:

为 zeromq 套接字设置 HWM。

增加 TCP 发送/接收缓冲区大小。

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

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

一个客户端一次最多可以发送 N 条消息。客户端的 RCVHWM 和代理的 SNDHWM 应该足够高以容纳总共 N 条消息。

之后，对于客户端收到的每个回复，它发送两个请求。