php - Ratchet PHP 和长时间运行的任务

Question

我正在使用 Ratchet php。我是这样开始的:

$loop    = \React\EventLoop\Factory::create();
$webSock = new \React\Socket\Server($loop);

$webSock->listen($this->port, $this->host);

$webServer = new \Ratchet\Server\IoServer(
    new \Ratchet\Http\HttpServer(
        new \Ratchet\WebSocket\WsServer(
            new PusherServer()
        )
    ),
    $webSock
);

return $loop;

现在，在我的 onMessage() 中我的 Pusherserver类(实现 MessageComponentInterface )，我想执行一个长时间的阻塞任务。这将是一个 HTTP 请求，最多可能需要 10 秒才能完成。

我如何制作 onMessage()在执行前一个 HTTP 请求时可以自由处理其他请求吗？我不能使用 pthreads，因为我无权更改我已经提供的 php 版本(这是线程安全的)。

Answer 1

这正是您在事件循环中执行任何操作时必须避免的问题:它不能被阻塞，因为其他任何人然后尝试订阅、调用消息或其他任何事件驱动的事件都不会发生，直到这就完成了。

这更像是一个架构问题，一旦你找到了最好的方法，那就是简化它并确保它适用于你需要的所有任务。

Ratchet 提供 ZMQ绑定(bind) - 这很棒，因为一旦你设置了它，你在端口 5555 上收到的任何东西都会在 onYourMethodName() 中命中你的事件循环。 ，或者随便你怎么称呼它!

考虑到这一点，您需要将需要完成的工作发送到另一个进程的作业队列中(react 有它的 child-process 扩展名，我特别不喜欢它，因为它在用户区进行轮询而不是中断 -驱动 I/O，如 PHP 的 PCNTL extension ) 或类似的。

如果您想“让它正常工作”，请启动需要完成的工作以及连接 ID 或其他 ID，以便您知道响应需要返回给谁，在子进程中以及何时完成发送出来。这不会阻塞!

如果你想以更好的方式去做，我强烈建议你研究一下这个和它的架构，这样当你再次处理这样的异步问题时，你可以把这些知识带在你的职业生涯中，采用'fire-and -忘记'的方法。将需要完成的工作触发到事件循环中的作业队列中，然后忘记它。

您的作业队列可以执行它的任务，当它完成时，通过 ZMQ(正在监听的端口 5555，记住)将结果返回，然后可以将数据发送回客户端。



对于关于工作队列的精彩演讲，我强烈推荐 this one来自 PHPNW。

最后要注意的是，因为你打开了这个东西并在端口 5555 上监听数据，所以你可以从任何地方发送这些数据。它可以是进程间通信，就像您有一个将数据发送到端口 5555 的 Java 应用程序一样，或者实际上是任何东西。它将事物绑定(bind)在一起，但不耦合它们，这在您的架构中很重要。

对于实际使用 ZMQ 的示例，他们在 this page here 上提供了所有内容。 (如上链接)，但我会尝试解释一下正在发生的事情。

$pull->bind('tcp://127.0.0.1:5555');
$pull->on('message', array($pusher, 'onYourMethodName'));

这部分意味着当任何东西向端口 5555 发送数据时，它是一个“消息”(你可以用谷歌搜索其他可用选项而不是消息)，它会调用 onYourMethodName在您的 $pusher目的。真的就是这么简单。超过 5555，点击 $pusher::onYourMethodName .

因此，您现在只需要在事件处理程序中创建您的方法(在 onMessage() 、 onSubscribe() 等旁边)......这在该页面上再次提到。

public function onYourMethodName($data) 
{
    /** You'll probably want to send the data in JSON format **/
    /** Imagine you get through a 'topic' in here... **/
    $data = json_decode($data, true);

    /** You should already have stored the people who are connected, topics etc - see the tutorial **/
    $topic = $this->subscribedTopics[$data['topic']];

    /** Send the data out to everyone subscribed to this topic **/
    $topic->broadcast($data);
}

如果您希望能够将数据发送给特定用户而不是所有人，那么有很多方法可以做到这一点。看看 this question对于我是如何做到的，但这是很久以前的事了。

您现在唯一需要自己做的事情是，在您的处理程序中(在 onMessage 或其他任何地方)，实际将需要完成的工作以及应将其发送回给谁(主题)。

在你的工作人员完成它的事情并获取数据的最后，它需要调用它来点击我上面显示的代码:

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

$socket->send(json_encode($data));

所以这是你需要做的:

开始监听 5555 端口

在您的事件处理程序等中创建您的方法

暂时不用担心排队或任何事情

创建一个非常简单的 php 脚本，执行上述 4 行代码，并证明当您单独运行该脚本时，它确实将数据发送到您的事件循环

然后担心如何将它添加到您的队列并让您的工作人员处理它