1

我正在使用新的Symfony Messenger Component 4.1 和RabbitMQ 3.6.10-1 从我的 Symfony 4.1 Web 应用程序中排队和异步发送电子邮件和 SMS 通知。我的 Messenger 配置 ( messenger.yaml) 如下所示：

framework:
    messenger:
        transports:
            amqp: '%env(MESSENGER_TRANSPORT_DSN_NOTIFICATIONS)%'

        routing:
            'App\NotificationBundle\Entity\NotificationQueueEntry': amqp

当要发送新通知时，我将其排队如下：

use Symfony\Component\Messenger\MessageBusInterface;
// ...
$notificationQueueEntry = new NotificationQueueEntry();
// [Set notification details such as recipients, subject, and message]
$this->messageBus->dispatch($notificationQueueEntry);

然后我在命令行上像这样启动消费者：

$ bin/console messenger:consume-messages

我已经实现了SendNotificationHandler实际交付发生的服务。服务配置：

App\NotificationBundle\MessageHandler\SendNotificationHandler:
    arguments:
        - '@App\NotificationBundle\Service\NotificationQueueService'
    tags: [ messenger.message_handler ]

和班级：

class SendNotificationHandler
{
    public function __invoke(NotificationQueueEntry $entry): void
    {
        $this->notificationQueueService->sendNotification($entry);
    }
}

到此为止，一切顺利，通知已送达。

现在我的问题是：由于（临时）网络故障，电子邮件或 SMS 可能无法发送。在这种情况下，我希望我的系统在指定的时间后重试交付，最多达到指定的最大重试次数。实现这一目标的方法是什么？

我已经阅读了关于Dead Letter Exchanges的文章，但是，我找不到任何关于如何将其与 Symfony Messenger 组件集成的文档或示例。

4

1 回答 1

3

你需要做的是告诉 RabbitMQ，消息被拒绝而不是被确认。默认情况下，Messenger 会在AmqpReceiver中处理这个问题。如您所见，如果您抛出实现RejectMessageExceptionInterface处理程序内部的异常，则该消息将自动为您拒绝。

您还可以使用自定义中间件“模拟”这种行为。我在一个小型演示应用程序中创建了类似的东西。该机制由一个中间件组成，该中间件将（序列化的）原始消息包装在一个新消息中RetryMessage，并通过自定义消息总线将其发送到不同的队列，用作死信交换。然后，该消息的处理程序将解包 RetryMessage（获取原始消息并对其进行反序列化）并通过默认总线传输它：

看：

这是拒绝消息并允许您立即再次使用它的基本设置（！）。您可能希望在延迟消耗时添加其他信息，例如时间戳的标头以对此进行改进。为此，您应该考虑编写自己的接收器、中间件和/或处理程序。

于 2018-10-30T17:40:15.683 回答