Есть несколько существующих вопросов, которые обсуждают, как использовать ZeroMQ, чтобы обойти возможность пропущенных сообщений, и большинство из них были очень поучительными.
Тем не менее, есть одна вещь, которая продолжает беспокоить меня из-за сокета ZMQ_DEALER. Я тестировал очень простой случай: 1 сервер и 2 клиента, каждый из которых использовал один сокет ZMQ_DEALER каждый. Сервер отправляет сообщения, а клиенты получают их.
Если сервер использует socket.bind () и client socket.connect (), мы можем наблюдать правильную циклическую балансировку, и уничтожение одного из клиентов приводит к тому, что сервер перенаправляет все свои сообщения оставшемуся клиенту. Без задержки, без потери пакетов, прекрасно работает.
Теперь, если у меня есть клиенты, выполняющие socket.bind () и сервер socket.connect () (все еще использующие один единственный сокет, но подключающиеся к обоим клиентам), это влияет на поведение сервера. После убийства одного из клиентов вместо перенаправления трафика на оставшийся, он будет продолжать балансировать нагрузку на обоих, пока количество сообщений в очереди не достигнет верхнего уровня для мертвого клиента.
Возможность использования connect на уже связанном сокете заставляет меня думать, что это будет более или менее симметричное использование, но мне было бы любопытно узнать причину такого поведения, и если есть способ воспроизвести переключение связанных сокетов на подключенные.
РЕДАКТИРОВАТЬ: для того, чтобы сделать вопрос немного более вдохновляющим, вот код для вас, чтобы проверить это поведение.
Это дилер:
// dealer.cc
// compile using something like this: g++ dealer.cc -o dealer -lzmq
#include <zmq.hpp>
#include <unistd.h>
#include <stdint.h>
int main() {
// prepare zmq
zmq::context_t context (1);
zmq::socket_t socket (context, ZMQ_DEALER);
socket.bind ("tcp://127.0.0.1:5555");
//socket.connect("tcp://127.0.0.1:5555");
//socket.connect("tcp://127.0.0.1:5556");
zmq::message_t msg;
int64_t more;
int counter = 0;
size_t more_size = sizeof more;
bool gotClients = false;
while (true) {
// send incrementing numbers
zmq::message_t world(sizeof(int));
memcpy(world.data(), &counter, sizeof(int));
socket.send(world);
counter++;
usleep(100000);
}
return 0;
}
А это клиент:
// client.cc
// compile using something like this: g++ client.cc -o client -lzmq
#include <zmq.hpp>
#include <iostream>
int main(int argc, char ** argv) {
zmq::context_t context (1);
zmq::socket_t socket (context, ZMQ_DEALER);
socket.connect("tcp://127.0.0.1:5555");
//socket.bind(argv[1]);
zmq::message_t msg;
while (true) {
socket.recv(&msg, 0);
std::cout << "Received a message: " << *(int *)msg.data() << std::endl;
}
return 0;
}
Создайте 2 клиентов, затем запустите дилера, затем убейте одного из клиентов, прежде чем убить дилера.
Если вы проверите вывод, то увидите, что ни одно сообщение не было отброшено или застряло в подвешенном состоянии очереди zmq. Нагрузка была сбалансирована, пока оба клиента были живы, и полностью перенаправлена на оставшегося, когда произошел «сбой».
Теперь давайте поменяем местами connect () / bind () для их инверсий (используя закомментированный код).
Мы должны сообщить клиентам, к какому адресу привязываться, чтобы они начали с URL следующим образом:
./client tcp://127.0.0.1:5555
./client tcp://127.0.0.1:5556
Затем, как и ранее, запустите дилера и убейте одного из клиентов.
Вы можете видеть, что оставшийся клиент получает только половину сообщений дилера, даже после того, как первый клиент был убит.
Насколько я понимаю, до тех пор, пока базовая очередь ZMQ не заполнена, дилер будет продолжать отправлять в очередь сообщения для отключенного однорангового узла (что в этом примере и с учетом параметров по умолчанию будет занимать действительно очень продолжительное время)