如何在独立代码中使用 Symfony Messenger 组件发送 AMQP 消息

Question

我们在 Symfony 5 项目中使用 Symfony Messenger 来与 RabbitMQ 集成。在 Symfony 中发送消息时它工作得很好，但我需要能够使用 Messenger 组件从一些不是使用 Symfony 框架构建的旧版 PHP 应用程序发送消息。

在 Symfony 下，它通过注入 MessageBusInterface 来处理所有的魔法，我需要做的就是这样：

    public function processTestMessage(MessageBusInterface $bus)

    {

        $bus->dispatch(new TestMessage('Hello World!');

    }

我需要以某种方式实例化我自己的版本，$bus它将以与 Symfony 相同的方式发送 AMQP 消息。我一直在尝试重新创建 Symfony 在幕后所做的一切来完成此任务，但无法将所有细节放在一起。

问题的关键是创建我自己的，SendMessageMiddleware与 Symfony 做同样的事情。之后，就很简单了：

    $sendersLocator = ???

    $eventDispatcher = ???

    $sendMessageMiddleware = new($sendersLocator, $eventDispatcher);

    $bus = new MessageBus([$sendMessageMiddleware]);

有没有人有使用 Messenger 组件在 Symfony 之外发送 AMQP 消息的工作代码示例？

Answer 1

这可以改进，但对我有用：use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpSender;use Symfony\Component\Messenger\Bridge\Amqp\Transport\Connection;use Symfony\Component\Messenger\Envelope;use Symfony\Component\Messenger\MessageBus;use Symfony\Component\Messenger\Middleware\SendMessageMiddleware;use Symfony\Component\Messenger\Transport\Sender\SendersLocatorInterface;$sendersLocator = new class implements SendersLocatorInterface {    public function getSenders(Envelope $envelope): iterable    {        $connection = new Connection(            [                'hosts' => 'localhost',                'port' => 5672,                'vhosts' => '/',                'login' => 'guest',                'password' => 'guest'            ],            [                'name' => 'messages'            ],            [                'messages' => []            ]        );        return [            'async' => new AmqpSender($connection)        ];    }};$middleware = new SendMessageMiddleware($sendersLocator);$bus = new MessageBus([$middleware]);$bus->dispatch(new MyMessage());

Answer 2

我修改了上面的答案，让我将 RabbitMQ 凭据作为环境变量传递。这就是我的应用程序所需要的。我试图编写自己的 DSN 解析器，发现 Symfony 已经做到了，所以我基本上从那里提取了代码。如果未设置环境变量，则默认使用上例中所示的相同设置。use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpSender;use Symfony\Component\Messenger\Bridge\Amqp\Transport\Connection;use Symfony\Component\Messenger\Envelope;use Symfony\Component\Messenger\MessageBus;use Symfony\Component\Messenger\Middleware\SendMessageMiddleware;use Symfony\Component\Messenger\Transport\Sender\SendersLocatorInterface;$sendersLocator = new class implements SendersLocatorInterface {    public function getSenders(Envelope $envelope): iterable    {        $dsn = getenv('MESSENGER_TRANSPORT_DSN') ?: $_ENV['MESSENGER_TRANSPORT_DSN'];        $connection = Connection::fromDsn($dsn);         return [            'async' => new AmqpSender($connection)        ];    }};$middleware = new SendMessageMiddleware($sendersLocator);$bus = new MessageBus([$middleware]);$bus->dispatch(new MyMessage());