Php , laravel, socket - 水平缩放技术

Question

我有一个 laravel 应用程序，我在其中使用 php ratchet 库在客户端和用户之间建立持久连接以进行实时通信(目前它只是一种方式，即，只有服务器可以发送消息到客户)。

有了这个架构，我如何拥有多个这样的 laravel 服务器，每个服务器都有自己的一组连接到它们的客户端，以便相互通信？

我正在考虑使用 REDIS 并拥有一个 redis 服务器，我的所有节点都将连接到该服务器(使用 pub sub)。

我的服务类似于群发消息应用程序(一个用户有很多组，一个组有很多用户)，到目前为止(现在我只有一个服务器，没有 redis)，我得到了所有用户的组，并为他订阅这些主题(主题名称类似于 GROUP_ 1、GROUP_ 8、GROUP_ 99 等)。

我想知道每个节点如何订阅 redis 中的特定主题(将所有 redis 消息发送到我的所有节点是没有意义的)。

Laravel 的示例说要像这样使用 php artisan 命令:

class RedisSubscribe extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'redis:subscribe';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = 'Subscribe to a Redis channel';

    /**
     * Execute the console command.
     *
     * @return mixed
     */
    public function handle()
    {
        Redis::subscribe(['test-channel'], function($message) {
            echo $message;
        });
    }
}

但是，我希望能够在运行时(运行 artisan 命令后)订阅、取消订阅某些主题。我怎样才能在 Laravel 中做到这一点？

Answer 1

1. 因此，如果我完全理解您的问题，那么您正在寻找一种方法来向连接在多个 WebsocketServer 上的用户发送消息。

2. 并且您不希望将其发送到没有订阅接收主题/ channel 的客户端的 websocketservers。

---

第 1 部分答案:

我目前还在大约 50 个不同的 websocketServer 上执行此操作，因此这是一个功能性设计。我将所有 50 个 websocketservers 的 IP 地址存储在我的数据库中。我还为每个 websocketservers 附加了 8096 个用户的限制。

发布/订阅部分

对于每个想要连接的用户，我首先调用 API 来确定用户必须连接到哪个 websocket。这是为了让一台服务器没有 8k 用户而其他服务器为空。实现此目的的另一种方法是使用负载平衡并在负载平衡器后面连接所有不同的 websocket 服务器。

完成后，我总是将用户连接到哪个 websocketserver 保存在我的数据库中。如果用户还订阅了 channel ，则会在数据库中添加一个新条目。

当用户取消订阅某个主题/ channel 时，该 channel 的用户条目将从数据库中删除。

发送消息

为了向不同的用户发送消息，我目前也在使用 API，因为我只使用从服务器到客户端的 websocket，而不是相反。

在用户向其他人发送消息时，API 调用还会注意到我正在运行的推送服务器。推送服务器也连接到数据库。

对于推送服务器收到的每条消息，它会及时了解哪些用户订阅了该 channel /主题。

然后它只会将数据发送到具有订阅了您正在寻址的 channel /主题的连接/用户的 websockets。

---

第 2 部分答案

它有点像以前的答案，但它的重点是只将数据发送到具有监听客户端的 websockets。此信息是从数据库中获取的。

我目前针对这个问题的类似数据库的方案(只包括其中的重要部分。

模板:表格(第 1 列，第 2 列)；

• 用户(用户 ID)；
• Websocket(websocketId, IP 地址, userLimit);
• 连接(connectionId, user.userId, websocket.websocketId);
• 订阅(connection.connectionId, channelId);

如果您需要更多信息，请告诉我。