我目前正在处理一个相当大的数据处理任务,需要拆分和分配处理.我有一个使用ZeroMQ的PUSH / PULL机制的完全工作的管道原型 – 正在发送“处理”数据并完成正常.
但是,一旦数据被扇出,并且处理完成,我需要指示工作者(在while / true循环中运行)终止.我认为使用ZMQ的PUB / SUB机制是明智的.但是,工作人员没有收到任何控制消息.
令人沮丧的是,如果我切换到PUSH / PULL机制,工作人员会收到消息并正确终止.
我在这里复制了原型中的一些代码：
您可以看到生产者(向工作人员发送数据的位)创建并绑定到控件套接字(第19行),工作者也会绑定(第53行).一旦从生产者完成工作,它通过控制套接字发送终止命令,工作人员继续循环并检查控制消息.不幸的是,第75行从未测试过,并且该过程继续循环.
// Control socket create / bind from producer
$this->controlSocket = $this->zmqContext->getSocket(ZMQ::SOCKET_PUB);
$this->controlSocket->bind('tcp://172.0.0.1:51001');
// Control socket create / bind from worker
$control = $this->getZmqContext()->getSocket(ZMQ::SOCKET_SUB);
$control->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, '');
$control->connect('tcp://172.0.0.1:51001');
解决方法:
是的,你可能会遇到“慢木匠”问题. PUB不会阻止,因此如果没有任何人发送消息,它将丢弃消息.因为您创建套接字并立即发送,订阅者仍将连接 – 这需要非零时间,因此消息会丢失.尝试在启动时立即创建控制套接字,应该给连接机会稳定(或者,在控制套接字中的bind()和send()之间放置一点睡眠).有关更详细的讨论,请参阅ZGuide：http://zguide.zeromq.org/page:all(搜索“慢速木匠”).
我注意到的另一件事是你在轮询之外检查控制套接字. poll的目的是让你做到这一点 – 将它作为另一个POLL_IN添加到你已经创建的pollset中,并检查read()中返回的socket是否为===(注意,三等于！)到控制插座或数据插座.这样你会得到更好的响应时间.