Symfony Messenger组件实战：构建支持多传输协议（AMQP/Redis）的异步消息总线 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Symfony Messenger 组件实战：构建支持多传输协议（AMQP/Redis）的异步消息总线

大家好，今天我们将深入探讨 Symfony Messenger 组件，并学习如何利用它构建一个支持多种传输协议（AMQP 和 Redis）的强大异步消息总线。我们将通过实际的代码示例，逐步了解 Messenger 的核心概念、配置方法以及如何根据业务需求选择合适的传输方式。

1. Messenger 组件简介

Symfony Messenger 是一个消息队列组件，它允许你在应用程序中异步处理任务。这意味着你可以将耗时的操作（例如发送电子邮件、处理图像、执行复杂的计算等）推送到消息队列，让它们在后台运行，而无需阻塞主请求。

Messenger 的核心概念包括：

Message (消息): 包含需要处理的数据的 PHP 对象。
Message Bus (消息总线): 接收消息并将其分发给合适的 Handler 的中心组件。
Handler (处理器): 包含实际处理消息逻辑的 PHP 类。
Transport (传输): 负责将消息从消息总线发送到消息队列，以及从消息队列接收消息。
Receiver (接收器): 监听消息队列，并将接收到的消息传递给消息总线。

2. 安装 Messenger 组件

首先，我们需要安装 Messenger 组件：

composer require symfony/messenger

3. 定义消息 (Message)

消息是一个简单的 PHP 对象，它包含了需要处理的数据。例如，假设我们需要发送欢迎邮件，我们可以定义一个 SendWelcomeEmailMessage：

<?php

namespace AppMessage;

class SendWelcomeEmailMessage
{
    private int $userId;

    public function __construct(int $userId)
    {
        $this->userId = $userId;
    }

    public function getUserId(): int
    {
        return $this->userId;
    }
}

4. 定义处理器 (Handler)

处理器负责处理消息。它是一个 PHP 类，包含一个 __invoke 方法，该方法接收消息作为参数。

<?php

namespace AppMessageHandler;

use AppMessageSendWelcomeEmailMessage;
use SymfonyComponentMailerMailerInterface;
use SymfonyComponentMimeEmail;
use SymfonyComponentDependencyInjectionAttributeAsMessageHandler;

#[AsMessageHandler]
class SendWelcomeEmailMessageHandler
{
    private MailerInterface $mailer;

    public function __construct(MailerInterface $mailer)
    {
        $this->mailer = $mailer;
    }

    public function __invoke(SendWelcomeEmailMessage $message)
    {
        // 从数据库中获取用户信息
        $user = $this->getUser($message->getUserId());

        // 构建邮件
        $email = (new Email())
            ->from('[email protected]')
            ->to($user->getEmail())
            ->subject('Welcome!')
            ->text('Welcome to our website!');

        // 发送邮件
        $this->mailer->send($email);

        // 记录日志或其他操作
        // ...
    }

    private function getUser(int $userId)
    {
        // 模拟从数据库获取用户
        return (object)['email' => 'user' . $userId . '@example.com'];
    }
}

5. 配置 Messenger

我们需要在 config/packages/messenger.yaml 文件中配置 Messenger。这个文件定义了消息总线、传输方式、路由规则等。

framework:
    messenger:
        transports:
            # AMQP 传输
            amqp: '%env(MESSENGER_TRANSPORT_DSN)%'

            # Redis 传输
            redis: '%env(MESSENGER_REDIS_DSN)%'

        routing:
            # 将 SendWelcomeEmailMessage 路由到 AMQP 传输
            AppMessageSendWelcomeEmailMessage: amqp

在这个配置中，我们定义了两个传输方式：amqp 和 redis。%env(MESSENGER_TRANSPORT_DSN)% 和 %env(MESSENGER_REDIS_DSN)% 是环境变量，用于指定 AMQP 和 Redis 服务器的 DSN。

路由规则将 AppMessageSendWelcomeEmailMessage 消息路由到 amqp 传输。这意味着当我们发送 SendWelcomeEmailMessage 消息时，它将被发送到 AMQP 消息队列。

6. 配置环境变量

我们需要在 .env 文件中配置 AMQP 和 Redis 的 DSN。

MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages
MESSENGER_REDIS_DSN=redis://localhost:6379/messages

确保你的 AMQP 和 Redis 服务器已经启动并运行。

7. 发送消息

现在我们可以发送消息了。在 Controller 或 Service 中，你可以注入 SymfonyComponentMessengerMessageBusInterface 并使用它的 dispatch 方法发送消息。

<?php

namespace AppController;

use AppMessageSendWelcomeEmailMessage;
use SymfonyBundleFrameworkBundleControllerAbstractController;
use SymfonyComponentHttpFoundationResponse;
use SymfonyComponentMessengerMessageBusInterface;
use SymfonyComponentRoutingAnnotationRoute;

class UserController extends AbstractController
{
    #[Route('/user/register/{id}', name: 'user_register')]
    public function register(int $id, MessageBusInterface $bus): Response
    {
        // 创建 SendWelcomeEmailMessage 消息
        $message = new SendWelcomeEmailMessage($id);

        // 发送消息
        $bus->dispatch($message);

        return new Response('User registered. Welcome email will be sent asynchronously.');
    }
}

8. 消费消息

我们需要运行 Messenger 的消费者来监听消息队列并处理消息。

php bin/console messenger:consume amqp

这个命令会启动一个消费者，它会监听 amqp 传输，并将接收到的消息传递给相应的处理器。

你也可以启动多个消费者，以提高消息处理的吞吐量。

9. 选择不同的传输方式

如果我们想将 SendWelcomeEmailMessage 消息路由到 Redis 传输，只需修改 config/packages/messenger.yaml 文件中的路由规则：

framework:
    messenger:
        transports:
            # AMQP 传输
            amqp: '%env(MESSENGER_TRANSPORT_DSN)%'

            # Redis 传输
            redis: '%env(MESSENGER_REDIS_DSN)%'

        routing:
            # 将 SendWelcomeEmailMessage 路由到 Redis 传输
            AppMessageSendWelcomeEmailMessage: redis

然后，你需要重启消费者，并指定 Redis 传输：

php bin/console messenger:consume redis

10. 优先级和重试机制

Messenger 支持优先级和重试机制，以确保重要消息得到及时处理，并在处理失败时进行重试。

优先级：

你可以在路由规则中指定消息的优先级。例如，将重要消息设置为 high 优先级：

framework:
    messenger:
        transports:
            # AMQP 传输
            amqp: '%env(MESSENGER_TRANSPORT_DSN)%'

            # Redis 传输
            redis: '%env(MESSENGER_REDIS_DSN)%'

        routing:
            # 将 SendWelcomeEmailMessage 路由到 AMQP 传输，并设置优先级
            AppMessageSendWelcomeEmailMessage:
                transport: amqp
                priority: high

重试机制：

Messenger 内置了重试机制，可以在消息处理失败时自动重试。你可以在 config/packages/messenger.yaml 文件中配置重试策略。

framework:
    messenger:
        failure_transport: failed

        transports:
            # AMQP 传输
            amqp: '%env(MESSENGER_TRANSPORT_DSN)%'

            # Redis 传输
            redis: '%env(MESSENGER_REDIS_DSN)%'

            failed: 'doctrine://default?queue_name=failed'

        routing:
            # 将 SendWelcomeEmailMessage 路由到 AMQP 传输
            AppMessageSendWelcomeEmailMessage: amqp

        retry_strategy:
            default:
                max_retries: 3
                delay: 1000 # 1 秒
                multiplier: 2
                max_delay: 3600000 # 1 小时

在这个配置中，我们定义了一个 failed 传输，用于存储处理失败的消息。retry_strategy 定义了默认的重试策略，包括最大重试次数、初始延迟、延迟倍增因子和最大延迟。

11. 使用 Doctrine Transport 存储失败消息

上述例子中，我们使用 Doctrine Transport 存储失败消息。首先，你需要安装 Doctrine Bridge：

composer require symfony/doctrine-messenger

然后，确保你的 Doctrine 配置正确，并创建一个用于存储失败消息的表：

php bin/console doctrine:schema:update --force

12. 事务性消息

如果你需要确保消息的发送和数据库操作在同一个事务中，可以使用 Messenger 的事务性消息功能。这可以避免消息发送成功但数据库操作失败，或者数据库操作成功但消息发送失败的情况。

首先，你需要配置 Doctrine 的事务性上下文：

framework:
    messenger:
        transports:
            # AMQP 传输
            amqp: '%env(MESSENGER_TRANSPORT_DSN)%'

            # Redis 传输
            redis: '%env(MESSENGER_REDIS_DSN)%'

        routing:
            # 将 SendWelcomeEmailMessage 路由到 AMQP 传输
            AppMessageSendWelcomeEmailMessage: amqp

        transactional:
            default:
                entity_manager: default

然后，在发送消息时，使用 SymfonyComponentMessengerEnvelope 类将消息包装起来，并传递给 MessageBusInterface 的 dispatch 方法。

<?php

namespace AppController;

use AppMessageSendWelcomeEmailMessage;
use DoctrineORMEntityManagerInterface;
use SymfonyBundleFrameworkBundleControllerAbstractController;
use SymfonyComponentHttpFoundationResponse;
use SymfonyComponentMessengerEnvelope;
use SymfonyComponentMessengerMessageBusInterface;
use SymfonyComponentRoutingAnnotationRoute;

class UserController extends AbstractController
{
    private EntityManagerInterface $entityManager;

    public function __construct(EntityManagerInterface $entityManager)
    {
        $this->entityManager = $entityManager;
    }

    #[Route('/user/register/{id}', name: 'user_register')]
    public function register(int $id, MessageBusInterface $bus): Response
    {
        // 创建 SendWelcomeEmailMessage 消息
        $message = new SendWelcomeEmailMessage($id);

        // 创建用户实体
        $user = new AppEntityUser();
        $user->setId($id);
        $user->setEmail('user' . $id . '@example.com');

        // 保存用户实体
        $this->entityManager->persist($user);
        $this->entityManager->flush();

        // 创建 Envelope 对象
        $envelope = new Envelope($message);

        // 发送消息
        $bus->dispatch($envelope);

        return new Response('User registered. Welcome email will be sent asynchronously.');
    }
}

在这个例子中，我们首先创建并保存了用户实体，然后创建了 Envelope 对象，并将消息传递给 dispatch 方法。Messenger 会自动将消息的发送和数据库操作放在同一个事务中。

13. 监控和调试

可以使用 Symfony Profiler 或其他监控工具来监控 Messenger 的性能和调试问题。Symfony Profiler 提供了 Messenger 的详细信息，包括消息的发送时间、处理时间、传输方式等。

总结

我们学习了如何使用 Symfony Messenger 组件构建一个支持多种传输协议（AMQP 和 Redis）的异步消息总线。我们了解了 Messenger 的核心概念、配置方法以及如何根据业务需求选择合适的传输方式。通过配置路由规则、优先级和重试机制，我们可以构建一个可靠且高效的消息队列系统。

选择合适的传输方式

AMQP 和 Redis 都是流行的消息队列系统，它们各有优缺点。

特性	AMQP (RabbitMQ)	Redis
数据持久性	支持持久化消息，即使服务器重启也不会丢失消息。	数据存储在内存中，持久化需要额外配置。
消息传递模式	支持多种消息传递模式，例如 Direct、Fanout、Topic、Headers。	支持简单的发布/订阅模式。
性能	适用于高吞吐量和低延迟的场景。	适用于高并发和快速响应的场景。
复杂性	配置和管理相对复杂。	配置和管理相对简单。
使用场景	适用于需要可靠性和复杂消息传递模式的场景，例如订单处理、支付系统。	适用于需要快速缓存和实时消息传递的场景，例如实时聊天、数据推送。

选择哪种传输方式取决于你的具体需求。如果需要可靠性和复杂的消息传递模式，AMQP 是一个不错的选择。如果需要快速缓存和实时消息传递，Redis 可能更适合。

配置调整和优化

根据实际情况，你可以调整 Messenger 的配置，以优化性能和资源利用率。例如，你可以调整消费者的数量、消息的优先级、重试策略等。

持续集成和部署

将 Messenger 集成到你的持续集成和部署流程中，以确保消息队列系统的稳定性和可靠性。你可以使用 Docker 等容器化技术来部署 AMQP 和 Redis 服务器，并使用自动化脚本来配置和管理 Messenger。

Symfony Messenger 组件实战：构建支持多传输协议（AMQP/Redis）的异步消息总线

发表回复 取消回复

发表回复取消回复