ZeroMQ SUBはメッセージを受信しません

Question

ZeroMQにPUB/SUBという問題があります。

すべてを接続した後、発行者は、すべてのメッセージを公開する（ソケットの送信メッセージがtrueを返す）が、SUBは.recv()機能に永遠に彼らとブロックを受け取ることはありません。ここで

は、私が使用していますコードです：

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?) 
} 

も試してみました（私は、2つの異なるスレッドでこれらの2つの機能をruningて最初SUBスレッドを開始、いくつかの時間を待った後、出版社のスレッドを開始していますのでご注意くださいそれ以外の方法では、エンドレスループでメッセージを送信するが、動作しなかった）。

私はここで間違っていますか？

Answer 1

あなたの例に基づいて、次のコードが私に役立ちます。 問題は、PUB/SUBパターンが遅い結合子であることです。つまり、PUBソケットをバインドしてメッセージを送信する前にしばらく待つ必要があります。

#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(); 
}