zeromq pub sub 上丢失的消息
Posted
技术标签:
【中文标题】zeromq pub sub 上丢失的消息【英文标题】:lost messages on zeromq 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;
【问题讨论】:
【参考方案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() 发送的消息没有已传输到网络的数据在关机时丢失。
要检查丢失了哪些消息,您可以尝试发送除时间戳之外的序列号。
【讨论】:
以上是关于zeromq pub sub 上丢失的消息的主要内容,如果未能解决你的问题,请参考以下文章