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. 您不希望通过将其发送到没有订阅接收主题/频道的客户端的 websocketservers 来产生开销。

---

第 1 部分答案：

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

发布/订阅部分

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

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

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

发送消息

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

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

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

然后它只会将数据发送到具有订阅该频道/主题的连接/用户的 websockets。

---

第 2 部分答案

这已经在前面的答案中了，但它的重点是只将数据发送到具有侦听客户端的 websocket。此信息是从数据库中获取的。

我目前针对此问题的类似数据库的方案（仅包括其中的重要部分。

模板： 表（第 1 列，第 2 列）；

• 用户(userId);
• Websocket(websocketId, IP-address, userLimit);
• 连接(connectionId, user.userId, websocket.websocketId);
• 订阅(connection.connectionId, channelId);

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