Zeromq代理只为每个请求提供了一个答复

Question

我现在正在开发具有ZMQ 4.3.1 C 绑定(CPPZMQ(的消息基础结构。我当前的架构是这样的：

    N Clients (Dealer) <------> [(Router) proxy (Dealer)] <------> N Workers (Dealer)

工作人员将处理请求并发送多个回复返回。

问题是我的客户端可以仅接收一个回复。客户线程接收第一个答复后，将在zmq::poll()上阻止。我已经测试了每个请求的一个答复非常完美，但是当工人寄回多个答复时，答复"消失了"，除了第一个。

代理代码：

_frontend = zmq::socket_t(_ctx, zmq::socket_type::router);
_backend = zmq::socket_t(_ctx, zmq::socket_type::dealer);
_controller.setsockopt(ZMQ_SUBSCRIBE, "", 0);
_frontend.setsockopt(ZMQ_SNDHWM, 0);
_frontend.setsockopt(ZMQ_RCVHWM, 0);
_backend.setsockopt(ZMQ_SNDHWM, 0);
_backend.setsockopt(ZMQ_RCVHWM, 0);
_frontend.bind("tcp://*:7777");
_backend.bind("tcp://*:7778");
zmq::proxy(_frontend, _backend);

客户端代码：

void client(int n) {
    zmq::context_t ctx(1);
    zmq::socket_t client(ctx, zmq::socket_type::dealer);
    client.setsockopt(ZMQ_RCVHWM, 0);
    std::string id = "Client-" + std::to_string(n);
    client.setsockopt(ZMQ_IDENTITY, id.c_str(), id.size());
    client.connect("tcp://localhost:7777");
    DataRequest request{utc_now(), utc_now(), "CL", {Timeframe::Five_Minute, Timeframe::Hourly, Timeframe::Daily, Timeframe::Weekly}};
    std::string req_str = request.serialize();
    zmq::message_t msg(req_str.c_str(), req_str.size());
    client.send(msg, zmq::send_flags::none);
    std::vector<zmq::pollitem_t> items = {
        {client, 0, ZMQ_POLLIN, 0}
    };
    while(true) {
        zmq::poll(items, -1);
        if(items[0].revents & ZMQ_POLLIN) {
            zmq::message_t rep;
            client.recv(rep, zmq::recv_flags::none);
            // process the message
        }
    }
}

服务器代码：

void server(int n) {
    // load data into memory
    std::vector<std::string> data;
    {
        // generate random message and push into data 
    }
    zmq::context_t ctx(1);
    zmq::socket_t server(ctx, zmq::socket_type::dealer);
    server.setsockopt(ZMQ_SNDHWM, 0);
    std::string id = "Server-" + std::to_string(n);
    server.setsockopt(ZMQ_IDENTITY, id.c_str(), id.size()); 
    server.connect("tcp://localhost:7778");
    std::vector<zmq::pollitem_t> items = {
        {server, 0, ZMQ_POLLIN, 0}
    };
    while(is_server_running) {
        zmq::poll(items, POLLER_TIMEOUT);
        if(items[0].revents & ZMQ_POLLIN) {
            zmq::message_t client_id, req;
            server.recv(client_id, zmq::recv_flags::none);
            server.recv(req, zmq::recv_flags::none);
            std::string client_addr = std::string(static_cast<char*>(client_id.data()), client_id.size()),
                        json_req = std::string(static_cast<char*>(req.data()), req.size());
            for(const auto& d: data) {
                zmq::message_t rep(d.c_str(), d.size());
                server.send(client_id, zmq::send_flags::sndmore);
                server.send(rep, zmq::send_flags::none);
                //std::this_thread::sleep_for(std::chrono::milliseconds(100));
            }
        }
    }
}

Answer 1

看这里，你会看到一个傻瓜...

最后，我解决了这个问题。发送后，client_id被移至Zeromq I/O队列中，因此它是以下答复的一个空消息对象。我仔细检查了Zeromq官方指南，发现这被称为"零拷贝"功能。

对于任何遇到类似问题的人，请提醒一开始检查身份。