zeromq pub sub 上丢失的消息答案

【问题标题】：lost messages on zeromq pub subzeromq pub sub 上丢失的消息
【发布时间】：2011-11-20 04:36:55
【问题描述】：

我正在尝试使用 zeromq 框架实现 pub sub 设计模式。这个想法是启动订阅者，然后启动发布者。订阅者将收听 100 条消息，发布者将发布 100 条消息。到目前为止，一切都很好... 然而，实际发生的情况是，即使在发布者启动时订阅者已经启动并运行，订阅者也不会收到所有消息（如果发布者发送至少 500 条消息，订阅者将收到 100 条消息）信息）。发布者发送的第一条消息似乎没有发送给订阅者。

有什么想法吗？

提前致谢，奥马尔。

订阅者代码（在发布者之前发布）

int i=0;
zmq::context_t context (1);
zmq::socket_t subscriber (context, ZMQ_SUB);
subscriber.connect("tcp://localhost:5556");
subscriber.setsockopt(ZMQ_SUBSCRIBE, "", 0);

for (int update_nbr = 0; update_nbr < 100; update_nbr++) 
{        
    zmq::message_t update;
    subscriber.recv(&update);
    i++;
    std::cout<<"receiving  :"<<i<<std::endl;
}

发布者代码（在订阅者之后发布）

zmq::context_t context (1);
zmq::socket_t publisher (context, ZMQ_PUB);
publisher.bind("tcp://*:5556");

int i = 0;
for (int update_nbr = 0; update_nbr < 100; update_nbr++) 
{        
    //  Send message to all subscribers
    zmq::message_t request (20);

    time_t seconds;
    seconds = time (NULL);

    char update [20]="";
    sprintf (update, "%ld", seconds);

    memcpy ((void *) request.data (), update,strlen(update));
    publisher.send(request);
    i++;
    std::cout << "sending :" << i << std::endl;

}

【问题讨论】：

标签： c++ publish-subscribe zeromq

【解决方案1】：

查看http://zguide.zeromq.org/page:all#Missing-Message-Problem-Solver 并在该网页上搜索“slow joiner”。

基本上，建立连接需要一点时间（几毫秒），在此期间可能会丢失大量消息。发布者需要在开始发布之前稍作休眠，或者（更好）它需要显式地与订阅者同步。

【讨论】：

我建议将潜艇与 pub 同步而不是直接睡眠......你永远不知道应该等待多长时间（尤其是如果你有远程潜艇）但是显式同步（通过 req/rep 套接字是一个好方法）你不必担心猜测......

【解决方案2】：

请查看guide。

发布者发送“你好”
每个收到“hello”的订阅者，通过 REQ/REP 套接字向发布者发送消息
当发布者获得足够的 REQ/REP 消息后，开始发布数据

【讨论】：

【解决方案3】：

在 0MQ 中，成功的 send() 并不意味着数据立即通过网络发送。 http://api.zeromq.org/2-1:zmq-send。您的消息非常小，AFAIR 0MQ 对小消息进行某种缓冲以更有效地使用网络。

如果我没记错的话，0MQ 的config.hpp 中的out_batch_size 控制着这种行为。

【讨论】：

但这并不能解释前几条消息丢失的原因。

【解决方案4】：

要查看的一件事（除了之前的评论者所指出的）是您的关机程序。

代码 sn-ps 可能只是不完整，但我看不出您是如何处理关机的。特别是您实际上可能会丢失发送的 last 消息。查看 zmq_close、zmq_term 和 ZMQ_LINGER 的文档。如果您没有实际调用这些函数，而只是简单地终止应用程序，那么有可能使用 zmq_send() 发送的消息没有已传输到网络的数据在关机时丢失。

要检查丢失了哪些消息，您可以尝试发送除时间戳之外的序列号。

【讨论】：