Symfony Messenger组件的定制化传输（Transport）：集成自定义消息队列

Symfony Messenger 组件：定制化传输，集成自定义消息队列

大家好，今天我们来深入探讨 Symfony Messenger 组件的定制化传输，以及如何集成自定义消息队列。Symfony Messenger 提供了一个强大的消息处理框架，允许我们在应用程序中解耦不同的组件，实现异步任务处理、事件驱动架构等。虽然 Messenger 默认支持 Doctrine、AMQP、Redis 等传输方式，但在实际项目中，我们可能会遇到需要集成特定的消息队列系统的情况。 这时候，定制化传输就显得尤为重要。

理解 Symfony Messenger 的核心概念

在深入定制化传输之前，我们需要先理解 Symfony Messenger 的几个核心概念：

• Message (消息): 应用程序中需要传递的数据，通常是一个 PHP 对象。
• Message Bus (消息总线): 消息的中心枢纽，负责接收消息并将其分发给相应的 Handler。
• Handler (处理器): 负责处理特定类型的消息，通常是一个实现了 MessageHandlerInterface 接口的类。
• Transport (传输): 负责将消息从 Message Bus 发送到消息队列，以及从消息队列接收消息并将其传递回 Message Bus。
• Serializer (序列化器): 负责将消息对象序列化为字符串（例如 JSON）以便存储在消息队列中，以及将字符串反序列化为消息对象。

为什么需要定制化传输？

Symfony Messenger 内置的传输方式已经足够强大，可以满足大多数场景的需求。然而，以下情况可能需要我们进行定制化传输：

• 集成未支持的消息队列系统: 例如，你需要集成公司内部自研的消息队列系统，或者某个小众但性能优异的消息队列产品。
• 特殊的消息格式要求: 某些消息队列系统可能对消息格式有特殊要求，例如需要使用特定的编码方式或协议。
• 自定义的错误处理逻辑: 你可能需要对消息队列的连接错误、消息发送失败等情况进行特殊处理，例如自动重试、报警等。
• 性能优化: 通过定制化传输，你可以针对特定的消息队列系统进行性能优化，例如批量发送消息、使用更高效的序列化方式等。

定制化传输的步骤

定制化传输主要涉及以下几个步骤：

1. 创建 Transport 类: 实现 TransportInterface 接口，负责消息的发送、接收、获取、确认和拒绝。
2. 创建 Connection 类: 负责与消息队列建立连接、发送消息、接收消息等操作。
3. 配置 Serializer: 根据消息队列的要求，选择合适的序列化器，或者自定义序列化器。
4. 配置 Messenger: 在 messenger.yaml 配置文件中配置自定义传输。

示例：集成一个简单的基于文件的消息队列

为了演示定制化传输，我们来创建一个简单的基于文件的消息队列，它将消息存储在文件中。请注意，这只是一个示例，不适用于生产环境。

1. 创建 Message 类:

<?php

namespace AppMessage;

class FileMessage
{
    private string $content;

    public function __construct(string $content)
    {
        $this->content = $content;
    }

    public function getContent(): string
    {
        return $this->content;
    }
}

2. 创建 FileConnection 类:

<?php

namespace AppTransport;

use SymfonyComponentMessengerExceptionTransportException;

class FileConnection
{
    private string $filePath;

    public function __construct(string $filePath)
    {
        $this->filePath = $filePath;
        if (!file_exists($filePath)) {
            touch($filePath);
        }
    }

    public function send(string $messageBody): void
    {
        try {
            file_put_contents($this->filePath, $messageBody . PHP_EOL, FILE_APPEND);
        } catch (Exception $e) {
            throw new TransportException(sprintf('Could not send message to file "%s": %s', $this->filePath, $e->getMessage()), 0, $e);
        }
    }

    public function get(): ?string
    {
        $messages = file($this->filePath, FILE_IGNORE_NEW_LINES | FILE_SKIP_EMPTY_LINES);
        if (empty($messages)) {
            return null;
        }

        $message = array_shift($messages);

        // 将剩余的消息写回文件
        file_put_contents($this->filePath, implode(PHP_EOL, $messages) . PHP_EOL);

        return $message;
    }

    public function ack(string $messageBody): void
    {
        // 在文件队列中，确认消息意味着从文件中删除消息。
        // 由于我们在 get() 方法中已经删除了消息，所以这里不需要做任何事情。
    }

    public function reject(string $messageBody): void
    {
        // 在文件队列中，拒绝消息意味着将消息放回队列。
        // 这里我们简单地将消息重新写入文件。
        $this->send($messageBody);
    }

    public function getFilePath(): string
    {
        return $this->filePath;
    }
}

3. 创建 FileTransport 类:

<?php

namespace AppTransport;

use SymfonyComponentMessengerEnvelope;
use SymfonyComponentMessengerExceptionTransportException;
use SymfonyComponentMessengerTransportSerializationSerializerInterface;
use SymfonyComponentMessengerTransportTransportInterface;

class FileTransport implements TransportInterface
{
    private FileConnection $connection;
    private SerializerInterface $serializer;

    public function __construct(FileConnection $connection, SerializerInterface $serializer)
    {
        $this->connection = $connection;
        $this->serializer = $serializer;
    }

    public function get(): iterable
    {
        $body = $this->connection->get();

        if (null === $body) {
            return [];
        }

        try {
            $envelope = $this->serializer->decode(['body' => $body, 'headers' => ['content-type' => 'application/json']]);

            return [$envelope];
        } catch (Exception $e) {
            throw new TransportException(sprintf('Could not decode message: %s', $e->getMessage()), 0, $e);
        }
    }

    public function ack(Envelope $envelope): void
    {
        $this->connection->ack($this->serializer->encode($envelope)->getBody());
    }

    public function reject(Envelope $envelope): void
    {
        $this->connection->reject($this->serializer->encode($envelope)->getBody());
    }

    public function send(Envelope $envelope): Envelope
    {
        try {
            $this->connection->send($this->serializer->encode($envelope)->getBody());

            return $envelope;
        } catch (Exception $e) {
            throw new TransportException(sprintf('Could not send message: %s', $e->getMessage()), 0, $e);
        }
    }
}

4. 创建 FileTransportFactory 类:

<?php

namespace AppTransport;

use SymfonyComponentMessengerTransportSerializationSerializerInterface;
use SymfonyComponentMessengerTransportTransportFactoryInterface;
use SymfonyComponentMessengerTransportTransportInterface;

class FileTransportFactory implements TransportFactoryInterface
{
    public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface
    {
        $filePath = substr($dsn, strlen('file://')); // Remove "file://" prefix

        return new FileTransport(new FileConnection($filePath), $serializer);
    }

    public function supports(string $dsn, array $options): bool
    {
        return str_starts_with($dsn, 'file://');
    }
}

5. 注册 FileTransportFactory 作为服务:

在 services.yaml 文件中注册 FileTransportFactory 服务：

services:
    AppTransportFileTransportFactory:
        tags:
            - { name: messenger.transport_factory }

6. 配置 Messenger:

在 messenger.yaml 文件中配置自定义传输：

framework:
    messenger:
        transports:
            file:
                dsn: 'file:///tmp/messages.txt'  # 文件路径
                serializer: messenger.transport.symfony_serializer
        routing:
            AppMessageFileMessage: file

7. 创建 Handler:

<?php

namespace AppMessageHandler;

use AppMessageFileMessage;
use SymfonyComponentMessengerHandlerMessageHandlerInterface;
use PsrLogLoggerInterface;

class FileMessageHandler implements MessageHandlerInterface
{
    private LoggerInterface $logger;

    public function __construct(LoggerInterface $logger)
    {
        $this->logger = $logger;
    }

    public function __invoke(FileMessage $message)
    {
        $this->logger->info('Received message: ' . $message->getContent());
    }
}

8. 发送消息:

<?php

namespace AppController;

use AppMessageFileMessage;
use SymfonyComponentHttpFoundationResponse;
use SymfonyComponentMessengerMessageBusInterface;
use SymfonyComponentRoutingAnnotationRoute;

class MessageController
{
    #[Route('/send', name: 'send_message')]
    public function sendMessage(MessageBusInterface $bus): Response
    {
        $message = new FileMessage('Hello from Symfony Messenger!');
        $bus->dispatch($message);

        return new Response('Message sent!');
    }
}

现在，当你访问 /send 路由时，一条包含 "Hello from Symfony Messenger!" 的消息将被发送到 /tmp/messages.txt 文件中。 FileMessageHandler 会从文件中读取这条消息并将其内容记录到日志中。

代码解析

• FileConnection: 负责与文件系统交互，将消息写入文件和从文件中读取消息。 send() 方法将消息追加到文件末尾。 get() 方法读取文件的第一行（即最早的消息），然后将剩余的消息写回文件，模拟队列的行为。 ack() 和 reject() 方法分别用于确认和拒绝消息。 在文件队列中，确认消息意味着从文件中删除消息，而拒绝消息意味着将消息放回队列。
• FileTransport: 实现了 TransportInterface 接口，负责消息的发送、接收、确认和拒绝。 它使用 FileConnection 与文件系统交互，并使用 SerializerInterface 将消息对象序列化为字符串，以及将字符串反序列化为消息对象。
• FileTransportFactory: 实现了 TransportFactoryInterface 接口，负责创建 FileTransport 对象。 supports() 方法用于判断给定的 DSN 是否支持 file:// 协议。 createTransport() 方法用于创建 FileTransport 对象，并将 FileConnection 和 SerializerInterface 注入到 FileTransport 中。
• services.yaml: 将 FileTransportFactory 注册为服务，并添加 messenger.transport_factory 标签，以便 Messenger 组件能够自动发现它。
• messenger.yaml: 配置 Messenger 组件，指定使用 file 传输，并设置 DSN 为 file:///tmp/messages.txt，指定文件路径。 routing 部分指定 AppMessageFileMessage 消息使用 file 传输。

进一步的优化和改进

以上示例只是一个简单的演示，实际项目中可能需要进行更多的优化和改进：

• 错误处理: 需要更完善的错误处理机制，例如重试机制、死信队列等。
• 并发处理: 文件队列在高并发场景下可能会出现竞争条件，需要使用锁机制来保证数据的一致性。
• 性能优化: 可以使用更高效的序列化方式，例如 MessagePack。 也可以考虑使用内存映射文件来提高读写性能。
• 监控和报警: 需要对消息队列的状态进行监控，例如队列长度、消息处理速度等。 当出现异常情况时，需要及时报警。
• 事务支持: 如果需要保证消息处理的原子性，需要使用事务。 例如，可以将消息的发送和数据库操作放在同一个事务中。

不同传输方式的比较

特性	Doctrine Transport	AMQP Transport	Redis Transport	自定义 Transport
适用场景	中小型项目，对性能要求不高	高并发，可靠性要求高	高性能，缓存场景	特定场景，定制需求
优点	简单易用	可靠性高，功能强大	性能高	灵活性高
缺点	性能较低	配置复杂	可靠性相对较低	开发成本高
是否需要扩展	通常不需要	某些特殊场景需要	某些特殊场景需要	必须
事务支持	支持	支持	不直接支持	根据实现而定
消息持久化	支持	支持	可以通过配置实现	根据实现而定
延迟队列支持	不支持	支持	支持	根据实现而定

总结：灵活运用定制化传输的强大功能

通过定制化传输，我们可以将 Symfony Messenger 组件集成到各种不同的消息队列系统中，满足各种各样的业务需求。 在设计定制化传输时，需要充分考虑消息队列的特性、性能要求、错误处理机制等因素，并进行充分的测试和验证，以确保系统的稳定性和可靠性。 定制化传输赋予了我们极大的灵活性，使我们能够构建更加强大和可扩展的应用程序。