8

我正在尝试使用 websockets 将一些数据从 php 应用程序发送到用户的浏览器。因此，我决定将Swoole与 RabbitMQ 结合使用。

这是我第一次使用 websockets，并且在阅读了一些关于 Socket.IO、Ratchet 等的帖子之后。我决定停止使用 Swoole，因为它是用 C 编写的并且可以方便地与 php 一起使用。

这就是我如何理解使用 websockets 启用数据传输的想法：1）在 CLI 中启动 RabbitMQ worker 和 Swoole 服务器 2）php 应用程序将数据发送到 RabbitMQ 3）RabbitMQ 将带有数据的消息发送到 worker 4）Worker 接收带有数据的消息 + 建立与 Swoole 套接字服务器的套接字连接。5) Swoole 服务器向所有连接广播数据

问题是如何将 Swoole 套接字服务器与 RabbitMQ 绑定？或者如何让 RabbitMQ 与 Swoole 建立连接并向其发送数据？

这是代码：

Swoole 服务器 (swoole_sever.php)

$server = new \swoole_websocket_server("0.0.0.0", 2345, SWOOLE_BASE);

$server->on('open', function(\Swoole\Websocket\Server $server, $req)
{
    echo "connection open: {$req->fd}\n";
});

$server->on('message', function($server, \Swoole\Websocket\Frame $frame)
{
    echo "received message: {$frame->data}\n";
    $server->push($frame->fd, json_encode(["hello", "world"]));
});

$server->on('close', function($server, $fd)
{
    echo "connection close: {$fd}\n";
});

$server->start();

Worker 从 RabbitMQ 接收消息，然后连接到 Swoole 并通过套接字连接（worker.php）广播消息

$connection = new AMQPStreamConnection('0.0.0.0', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('task_queue', false, true, false, false);

echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";

$callback = function($msg){
    echo " [x] Received ", $msg->body, "\n";
    sleep(substr_count($msg->body, '.'));
    echo " [x] Done", "\n";
    $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);


    // Here I'm trying to make connection to Swoole server and sernd data
    $cli = new \swoole_http_client('0.0.0.0', 2345);

    $cli->on('message', function ($_cli, $frame) {
        var_dump($frame);
    });

    $cli->upgrade('/', function($cli)
    {
        $cli->push('This is the message to send to Swoole server');
        $cli->close();
    });
};

$channel->basic_qos(null, 1, null);
$channel->basic_consume('task_queue', '', false, false, false, false, $callback);

while(count($channel->callbacks)) {
    $channel->wait();
}

$channel->close();
$connection->close();

消息将发送到 RabbitMQ (new_task.php) 的新任务：

$connection = new AMQPStreamConnection('0.0.0.0', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('task_queue', false, true, false, false);

$data = implode(' ', array_slice($argv, 1));
if(empty($data)) $data = "Hello World!";
$msg = new AMQPMessage($data,
    array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
);

$channel->basic_publish($msg, '', 'task_queue');

echo " [x] Sent ", $data, "\n";

$channel->close();
$connection->close();

在启动 swoole 服务器和 worker 之后，我正在从命令行触发 new_task.php：

php new_task.php

在运行 RabbitMQ Worker (worker.php) 的命令行提示符中，我可以看到一条消息已传递给 worker（“[x] Received Hello World!”消息正在出现）。

然而，在运行 Swoole 服务器的命令行提示符中什么也没有发生。

所以问题是：1）这种方法的想法正确吗？2）我做错了什么？

4

1 回答 1

1

在收到消息时触发的回调（in worker.php）中，您使用swoole_http_client的是仅异步的。这似乎导致代码永远不会被完全执行，因为回调函数在异步代码被触发之前返回。

做同样事情的同步方法将解决这个问题。这是一个简单的例子：

$client = new WebSocketClient('0.0.0.0', 2345);
$client->connect();
$client->send('This is the message to send to Swoole server');
$recv = $client->recv();
print_r($recv);
$client->close();

查看github 上WebSocketClient的类和示例用法。

您也可以将其包装在coroutine中，如下所示：

go(function () {
    $client = new WebSocketClient('0.0.0.0', 2345);
    $client->connect();
    $client->send('This is the message to send to Swoole server');
    $recv = $client->recv();
    print_r($recv);
    $client->close();
});
于 2018-05-21T19:46:59.237 回答