ZeroMQ SUB empfängt nie Nachrichten

Question

Ich habe Probleme mit einem PUB/SUB in ZeroMQ.

Nachdem alles verbindet, veröffentlicht Publisher alle Nachrichten (Buchse des Nachrichtens kehrt true), aber die SUB nie empfängt sie und sperrt für immer auf .recv() Funktion.

Hier ist der Code ich verwende:

void startPublisher() 
{ 
    zmq::context_t zmq_context(1); 
    zmq::socket_t zmq_socket(zmq_context, ZMQ_PUB); 
    zmq_socket.bind("tcp://127.0.0.1:58951"); 

    zmq::message_t msg(3); 
    memcpy(msg.data(), "abc", 3); 

    for(int i = 0; i < 10; i++) 
     zmq_socket.send(msg); // <-- always true 
} 

void startSubscriber() 
{ 
    zmq::context_t zmq_context(1); 
    zmq::socket_t zmq_socket(zmq_context, ZMQ_SUB); 

    zmq_socket.connect("tcp://127.0.0.1:58951"); 
    zmq_socket.setsockopt(ZMQ_SUBSCRIBE, "", 0); // allow all messages 

    zmq::message_t msg(3); 
    zmq_socket.recv(&msg); // <-- blocks forever (message never received?) 
} 

Bitte beachten Sie, dass ich diese zwei Funktionen in zwei verschiedenen Threads bin runing, beginnend SUB Faden zunächst für einige Zeit warten und dann Verleger Thread starten (auch versucht, andersherum mit dem Verleger Senden von Nachrichten in einer Endlosschleife, hat aber nicht funktioniert).

Was mache ich hier falsch?

Answer 1

Basierend auf Ihrem Beispiel funktioniert der folgende Code für mich. Das Problem ist, dass das PUB/SUB-Pattern ein langsamer Joiner ist. Das bedeutet, dass Sie nach dem Binden des PUB-Sockels und vor dem Senden einer Nachricht etwas warten müssen.

#include <thread> 
#include <zmq.hpp> 
#include <iostream> 
#include <unistd.h> 
void startPublisher() 
{ 
    zmq::context_t zmq_context(1); 
    zmq::socket_t zmq_socket(zmq_context, ZMQ_PUB); 
    zmq_socket.bind("tcp://127.0.0.1:58951"); 
    usleep(100000); // Sending message too fast after connexion will result in dropped message 
    zmq::message_t msg(3); 
    for(int i = 0; i < 10; i++) { 
     memcpy(msg.data(), "abc", 3); 
     zmq_socket.send(msg); // <-- always true 
     msg.rebuild(3); 
     usleep(1); // Temporisation between message; not necessary 
    } 
} 
volatile bool run = false; 
void startSubscriber() 
{ 
    zmq::context_t zmq_context(1); 
    zmq::socket_t zmq_socket(zmq_context, ZMQ_SUB); 
    zmq_socket.connect("tcp://127.0.0.1:58951"); 
    std::string TOPIC = ""; 
    zmq_socket.setsockopt(ZMQ_SUBSCRIBE, TOPIC.c_str(), TOPIC.length()); // allow all messages 
    zmq_socket.setsockopt(ZMQ_RCVTIMEO, 1000); // Timeout to get out of the while loop 
    while(run) { 
     zmq::message_t msg; 
     int rc = zmq_socket.recv(&msg); // Works fine 
     if(rc) // Do no print trace when recv return from timeout 
      std::cout << std::string(static_cast<char*>(msg.data()), msg.size()) << std::endl; 
    } 
} 
int main() { 
    run = true; 
    std::thread t_sub(startSubscriber); 
    sleep(1); // Slow joiner in ZMQ PUB/SUB pattern 
    std::thread t_pub(startPublisher); 
    t_pub.join(); 
    sleep(1); 
    run = false; 
    t_sub.join(); 
}