PUSH/PULL 模式中丢失的消息（Ratchet + PHP + ZeroMQ 推送集成）

Posted 2023-02-16

【中文标题】PUSH/PULL 模式中丢失的消息（Ratchet + PHP + ZeroMQ 推送集成）

【英文标题】：Lost messages in PUSH/PULL pattern ( Ratchet + PHP + ZeroMQ push integration )

【发布时间】：2017-09-28 15:39:09

【问题描述】：

我正在我的网站上创建聊天、推送通知系统、用户活动小部件（即时更新）等。

我的网站是基于 php 构建的，所以我决定使用 Ratchet 作为我的任务的 websocket 服务器。我已经安装了所有必需的组件，并学习了http://socketo.me/docs/push 上的指南并开始编写代码。

这是在 model.php 文件中的 ChatMsg( $item )... 方法中。它创建一个 PUSH 套接字访问点原型，并在将新项目插入数据库后通过 ZeroMQ 向服务器发送带有 JSON 数据的消息：

$context = new ZMQContext();
$socket = $context->getSocket(ZMQ::SOCKET_PUSH, 'my pusher');
$socket->connect("tcp://localhost:5555");
$socket->send(json_encode($sData));

接下来是我的push-server.php，只创建一个PULL套接字接入点原型并等待新消息，这些消息将被传输到推送脚本，向客户广播新的通知、聊天消息和其他事件。

<?php 
require dirname(__DIR__) . '/vendor/autoload.php';

    $loop   = React\EventLoop\Factory::create();
    $pusher = new MyApp\Pusher;

    // Listen for the web server to make a ZeroMQ push after an ajax request
    $context = new React\ZMQ\Context($loop);
    $pull = $context->getSocket(ZMQ::SOCKET_PULL);
    $pull->setSockOpt(ZMQ::SOCKOPT_HWM, 0);
    $pull->bind('tcp://127.0.0.1:5555'); // Binding to 127.0.0.1 means the only client that can connect is itself
    $pull->on('error', function ($e) 
        $f = fopen('push-server-error.log', "a");
        fwrite($f, $e->getMessage()."\n");
        fclose($f);
    );
    $pull->on('message', array($pusher, 'onNewEvent'));

    // Set up our WebSocket server for clients wanting real-time updates
    $webSock = new React\Socket\Server($loop);
    $webSock->listen(8081, '0.0.0.0'); // Binding to 0.0.0.0 means remotes can connect
    $webServer = new Ratchet\Server\ioserver(
        new Ratchet\Http\HttpServer(
            new Ratchet\WebSocket\WsServer(
                new Ratchet\Wamp\WampServer(
                    $pusher
                )
            )
        ),
        $webSock
    );

    $loop->run();
?>

我使用监控工具 Supervisor 成功启动了 push-server.php，我为 WebSocket 流量设置了 nginx 代理，设置了客户端脚本（高速公路等）。

一般来说，我打算在生产中使用它。最初几个小时，我在自己的网站上修改了新的聊天系统，我对其进行了测试，一切都运行良好。

但后来我遇到了这个问题。通过 ZeroMQ PUSH 套接字发送后，一些 ZeroMQ 消息（仅其中一部分，可能 5-10%）丢失。那时，自 push-server.php 进程启动以来，当发送了大约 300-400 条消息时，就会出现此问题。

我深信这个问题在 ZeroMQ 内部（不在 JS 客户端或带有业务逻辑的 Pusher 脚本内部），因为我试图修改 ->on()... 中的方法>push-server.php 以便在终端（控制台）上显示新消息，丢失的消息甚至不显示在控制台上，即“->on()...”方法没有赶上它们。

ZeroMQ "->send()" 方法总是在消息发送成功或丢失时返回一个空的 ZeroMQ 套接字对象。我只是通过在我的网站上发送聊天消息并获得响应（通过 AJAX 实现表单提交）来检查这一点：

var_dump($socket->send(json_encode($sData)));

会出现什么问题以及如何解决？

Server OS:      CentOS 6.9 (Final)
PHP version:           5.6.31
ZMQ extension version: 1.1.3
libzmq version:        4.2.2

【问题讨论】：

欢迎来到 Stack Overflow。请阅读***.com/help/mcve。尝试隔离问题，以使用最少的组件可靠地重现它。既然您深信这个问题在 ZMQ 内部，那么从那里开始是有意义的，从方程式中排除 Ratchet 和前端。

【参考方案1】：

我创建了 not persistent ZMQ::Context 和 ZMQ::Socket 并且我的问题得到了解决：

$context = new ZMQContext(1, false);
$socket = $context->getSocket(ZMQ::SOCKET_PUSH);

【讨论】：

您好，您对这如何解决您的问题有任何见解吗？您丢失消息是因为它是一个持久队列吗？