php - 在阻塞应用程序中使用响应式 PHP

Question

我目前正在开发一个 PHP 应用程序，它将使用一些 websocket 连接来与另一个服务通信。

为了与这个 websocket 服务通信，我们使用 Ratchet - 这是一个基于 react PHP 的 PHP 库。

这段代码需要发送和响应几个请求，然后将信息返回给“主线程”。

示例流程:

HTTP 请求 -> Controller -> 启动打开 websocket 客户端的服务 -> websocket 客户端正在与服务器通信 -> 一旦完成，它应该将结果返回给 Controller 代码 -> Controller 输出给用户

我遇到的问题是我不熟悉 Reactive PHP 并且不知道如何处理这个问题。

我试过了;

    $service = new WebsocketService();
    $startTimer = time();
    $service->getList(44);
    while($service->getResponse() == null) {
        usleep(500);
        if (time() > $startTimer + 10) {
            continue; //Timeout on 10 seconds
        }
    }
    var_dump($service->getResponse());

一旦完成，服务代码会将其“响应”变量设置为 null 以外的值。这显然失败了，因为 sleep 方法阻塞了线程。同样没有，似乎 while 循环阻塞了 I/O 并且 react 代码失败了。

一个解决方案是打开一个新线程并在那里运行 websocket 代码，但我不会对此感到满意。

我觉得我需要围绕 websocket 进程实现某种“观察者”，但我不知道该怎么做。

我们的 Websocket 服务客户端代码如下所示；

private $response = null;

/**
 * @return null|object
 */
public function getResponse() {
    return $this->response;
}

public function getList($accountId) {
    $this->response = null;
    \Ratchet\Client\connect('ws://192.168.56.1:8080')->then(function(\Ratchet\Client\WebSocket $conn) use ($accountId) {
        $login = new \stdClass();
        $login->action = 'login';
        $conn->on('message', function($msg) use ($conn, $login, $accountId) {
            try {
                $response = json_decode($msg);
                if ($response->result_id == 100) {
                    //Succesfully logged in to websocket server

                    //Do our request now.
                    $message = new \stdClass();
                    $message->target = 'test';
                    $conn->send(json_encode($message));
                }

                if (isset($response->reply) && $response->reply == 'list') {
                    $this->response = $response; //This is the content I need returned in the controller
                    $conn->close(); //Dont need it anymore
                }

            } catch (\Exception $e) {
                echo 'response exception!';
                //Do nothing for now
            }
        });

        $conn->send(json_encode($login));
    }, function ($e) {
        echo "Could not connect: {$e->getMessage()}\n";
    });
}

像这样运行代码也不起作用；

    $service = new WebsocketService();
    $service->getList(44);
    echo 'Test';
    var_dump($service->getResponse());

因为“测试” echo 出现在我什至从 websocket 服务器得到响应之前。

请赐教!我不确定要搜索什么。

Answer 1

PHP 和 websockets 似乎仍然有点实验性。尽管如此，我在 medium.com 上找到了一个很棒的教程，由 Adam Winnipass 编写，它应该对解决您的问题很有帮助:https://medium.com/@winni4eva/php-websockets-with-ratchet-5e76bacd7548

唯一的区别是他们使用 JavaScript 而不是 PHP 来实现他们的 websocket 客户端。但最终应该不会有太大区别，因为一旦我们打开了每一端的 Websocket 连接，两个应用程序都必须发送并等待接收通知 - 这就是他们说明它的方式:



似乎创建成功的 Websocket 连接的一种可能性是扩展 MessageComponentInterface

use Ratchet\MessageComponentInterface;

这也需要

use Ratchet\ConnectionInterface;

消息组件接口(interface)定义了以下方法:

打开

在线留言

关闭

错误

我认为这就是 Ratchet 库的实现方式。这就是他们最终启动服务器的方式:

use Ratchet\Server\IoServer;
use MyApp\MyCustomMessageComponentInterface;
use Ratchet\Http\HttpServer;
use Ratchet\WebSocket\WsServer;
require dirname(__DIR__) . '/vendor/autoload.php';
$server = IoServer::factory(
            new HttpServer(
                new WsServer(
                    new MyCustomMessageComponentInterface()
                )
            ),
          8080
         );
$server->run();

使用这种架构，您已经可以接收 (onMessage) 并且使用 send() 方法也可以发送。

我无法用您现有的代码解决确切的问题。但我想如果你是 按预期使用库的预构建类和接口(interface) (并在此处演示)您应该能够通过将代码添加到相应的方法来实现您想要的。

更多信息和示例可以在文档中找到:

http://socketo.me/docs/server

http://socketo.me/api/namespace-Ratchet.html