告别PubSub消息处理的泥潭：如何使用gos/pubsub-router-bundle优雅地管理异步逻辑

<code class="bash">composer require gos/pubsub-router-bundle</code>

<pre class="brush:php;toolbar:false;">// app/AppKernel.php
class AppKernel extends Kernel
{
    public function registerBundles()
    {
        $bundles = array(
            // ... 其他 bundles
            new \Gos\Bundle\PubSubRouterBundle\GosPubSubRouterBundle()
        );

        // ...
    }
}

<pre class="brush:php;toolbar:false;"># gos_pubsub_router.yaml
gos_pubsub_router:
    routers:
        websocket: # 定义一个名为 'websocket' 的路由器
            resources:
                - '%kernel.project_dir%/config/pubsub/websocket_routes.yaml' # 路由定义文件路径
        redis: # 定义一个名为 'redis' 的路由器
            resources:
                - '@AppBundle/Resources/config/pubsub/redis_routes.yaml' # 另一个路由定义文件路径

<pre class="brush:php;toolbar:false;"># config/pubsub/websocket_routes.yaml
user_notification_channel: # 路由名称
    channel: notification/user/{role}/{application}/{user_ref} # 频道模式，支持占位符
    handler: ['App\MessageHandler\NotificationHandler', 'handleUserMessage'] # 处理器：可以是服务、类方法或PHP函数
    requirements: # 占位符的正则表达式要求
        role: "editor|admin|client"
        application: "[a-z]+"
        user_ref: "\d+"

chat_room_message:
    channel: chat/room/{roomId}/message/{userId}
    handler: 'App\Service\ChatService::processRoomMessage' # 也可以直接是服务方法字符串
    requirements:
        roomId: "\d+"
        userId: "\d+"

<pre class="brush:php;toolbar:false;"><?php

namespace App\Service;

use Gos\Bundle\PubSubRouterBundle\Generator\GeneratorInterface;

class PublisherService
{
    private GeneratorInterface $websocketRouter;

    public function __construct(
        #[AsService(id: 'gos_pubsub_router.websocket')] GeneratorInterface $websocketRouter
    ) {
        $this->websocketRouter = $websocketRouter;
    }

    public function publishNewUserMessage(string $role, string $application, int $userId, string $message): void
    {
        $channel = $this->websocketRouter->generate('user_notification_channel', [
            'role' => $role,
            'application' => $application,
            'user_ref' => $userId,
        ]);

        // 假设你有一个消息发布客户端
        // $this->messageClient->publish($channel, $message);

        echo "Generated channel: " . $channel . "\n";
        // 示例输出: Generated channel: notification/user/admin/blog-app/123
    }
}

<pre class="brush:php;toolbar:false;"><?php

namespace App\Service;

use Gos\Bundle\PubSubRouterBundle\Matcher\MatcherInterface;
use Gos\Bundle\PubSubRouterBundle\Exception\ResourceNotFoundException;
use Symfony\Component\DependencyInjection\ContainerInterface; // 仅为示例，实际应避免直接使用容器

class MessageProcessor
{
    private MatcherInterface $websocketRouter;
    private ContainerInterface $container; // 用于获取处理器服务

    public function __construct(
        #[AsService(id: 'gos_pubsub_router.websocket')] MatcherInterface $websocketRouter,
        ContainerInterface $container
    ) {
        $this->websocketRouter = $websocketRouter;
        $this->container = $container;
    }

    public function processIncomingMessage(string $channel, string $payload): void
    {
        try {
            list($routeName, $route, $attributes) = $this->websocketRouter->match($channel);

            echo "Matched route: " . $routeName . "\n";
            echo "Extracted attributes: " . json_encode($attributes) . "\n";

            // 获取并调用处理器
            $handlerConfig = $route->getHandler();
            if (is_array($handlerConfig) && count($handlerConfig) === 2) {
                $serviceId = $handlerConfig[0];
                $method = $handlerConfig[1];

                $handlerService = $this->container->get($serviceId); // 从容器获取服务
                $handlerService->$method($attributes, $payload); // 调用处理器方法，传入参数和消息体
            } elseif (is_string($handlerConfig) && str_contains($handlerConfig, '::')) {
                list($serviceId, $method) = explode('::', $handlerConfig);
                $handlerService = $this->container->get($serviceId);
                $handlerService->$method($attributes, $payload);
            } else {
                // 处理其他类型的处理器，例如全局函数
                echo "Unsupported handler type for route: " . $routeName . "\n";
            }

        } catch (ResourceNotFoundException $e) {
            echo "No route found for channel: " . $channel . "\n";
            // 记录日志或处理未匹配的频道
        }
    }
}

// 假设你的 NotificationHandler 服务
namespace App\MessageHandler;

class NotificationHandler
{
    public function handleUserMessage(array $attributes, string $payload): void
    {
        echo sprintf(
            "Handling user message for role '%s', app '%s', user '%s'. Payload: %s\n",
            $attributes['role'],
            $attributes['application'],
            $attributes['user_ref'],
            $payload
        );
        // 这里是你的业务逻辑，例如发送推送通知
    }
}

// 模拟调用
$messageProcessor->processIncomingMessage('notification/user/admin/blog-app/123', 'Hello Admin!');
// 输出:
// Matched route: user_notification_channel
// Extracted attributes: {"role":"admin","application":"blog-app","user_ref":"123"}
// Handling user message for role 'admin', app 'blog-app', user '123'. Payload: Hello Admin!

$messageProcessor->processIncomingMessage('chat/room/101/message/200', 'What\'s up?');
// 输出:
// Matched route: chat_room_message
// Extracted attributes: {"roomId":"101","userId":"200"}
// ... (ChatService::processRoomMessage 被调用)

$messageProcessor->processIncomingMessage('unknown/channel/123', 'Test');
// 输出: No route found for channel: unknown/channel/123

<code class="bash">php bin/console gos:prouter:debug -r websocket</code>