Zeromq прокси доставил только один ответ на каждый запрос - PullRequest
Новая
       17

Zeromq прокси доставил только один ответ на каждый запрос

1 голос
/ 28 июня 2019

Сейчас я разрабатываю инфраструктуру обмена сообщениями с привязкой 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));
            }
        }
    }
}

1 Ответ

1 голос
/ 28 июня 2019

Посмотрите сюда, и вы увидите дурака ...

Наконец-то я решил эту проблему.client_id перемещается в очередь ввода / вывода ZeroMQ после отправки, поэтому он является пустым объектом сообщения для следующих ответов.Я дважды проверил официальное руководство ZeroMQ и обнаружил, что это называется функцией «Zero-Copy».

Для тех, кто сталкивался с подобной проблемой, вам следует сначала проверить IDENTITY.

...