php - ZeroMQ PUB/SUB 不适用于 PHP

Question

我目前正在处理一个相当大的数据处理任务，需要拆分和分配处理。我有一个使用 ZeroMQ 的 PUSH/PULL 机制的完整工作流水线原型(prototype) - 正在发送“处理”数据并完成 OK。

但是，一旦数据被扇出并且处理完成，我需要指示工作人员(在 while/true 循环中运行)终止。我认为为此使用 ZMQ 的 PUB/SUB 机制是明智的。但是，工作人员没有收到任何控制消息。

令人沮丧的是，如果我切换到 PUSH/PULL 机制，工作人员会收到消息并正确终止。

我从这里复制了一些原型(prototype)代码:

http://pastebin.com/myZyWQ1E

您可以看到生产者(向工作人员发送数据的位)创建并绑定(bind)到控制套接字(第 19 行)，工作人员也绑定(bind)该控制套接字(第 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');

Answer 1

是的，您可能遇到了“慢加入者”问题的一个实例。 PUB 不会阻塞，因此如果没有人可以将消息发送给它，它将丢弃消息。因为您创建了套接字并立即发送，订阅者仍将保持连接——这需要非零时间，所以消息丢失了。一开始就尝试创建控制套接字，应该让连接有机会稳定下来(或者，在控制套接字中的 bind() 和 send() 之间稍作休息)。有关更详细的讨论，请参阅 ZGuide:http://zguide.zeromq.org/page:all (搜索“慢木匠”)。

我注意到的另一件事是您正在轮询之外检查控制套接字。 poll 的目的就是让你做那种事情 - 将它作为另一个 POLL_IN 添加到你已经创建的 pollset 并检查 read() 中返回的套接字是否 === (注意，三重等于!)到控制套接字或数据套接字。这样你会得到更好的响应时间。