2

我们正在使用 Symfony Messenger，并且有这些传输：

framework:
    messenger:
        failure_transport: failed

        transports:
            failed:
                dsn: 'doctrine://default?queue_name=failed'
                options:
                    table_name: 'MessengerMessages'
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    max_retries: 3
                    delay: 5000
                    multiplier: 2
                    max_delay: 0
            asyncLowPriority:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%_low_priority'
                retry_strategy:
                    max_retries: 5
                    delay: 3600000
                    multiplier: 2
                    max_delay: 0
            sync: 'sync://'

当我们向队列发送消息async，并且最后一次重试失败并出现异常时，异常被记录到MessengerMessages表中，并且异常冒泡（在我们的例子中进入Sentry）。这就是我们想要的。

然而，当我们向队列发送消息asyncLowPriority时，我们希望失败的消息发送到：

  • 没到failed交通工具
  • 不要让异常冒泡

基本上，应该删除异常。

这可能吗，怎么做？

原因是我们使用这个队列来异步下载图像，并且我们已经在命令处理程序的专用数据库表中记录了每个失败。

4

1 回答 1

1

我设法用中间件做到了这一点：


use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Exception\HandlerFailedException;
use Symfony\Component\Messenger\Exception\UnrecoverableMessageHandlingException;
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
use Symfony\Component\Messenger\Middleware\StackInterface;
use Symfony\Component\Messenger\Stamp\ReceivedStamp;
use Symfony\Component\Messenger\Stamp\RedeliveryStamp;
use Symfony\Component\Messenger\Stamp\SentToFailureTransportStamp;
use Throwable;

final class BypassFailureTransportMiddleware implements MiddlewareInterface
{
    public function __construct(
        private string $transportName,
        private int $maxRetries,
    ) {
    }

    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        try {
            return $stack->next()->handle($envelope, $stack);
        } catch (HandlerFailedException $exception) {
            $nestedException = $this->getNestedException($exception);

            if ($nestedException === null) {
                throw $exception;
            }

            /** @var ReceivedStamp|null $receivedStamp */
            $receivedStamp = $envelope->last(ReceivedStamp::class);

            if ($receivedStamp === null || $receivedStamp->getTransportName() !== $this->transportName) {
                throw $exception;
            }

            if (!$this->isLastRetry($envelope, $nestedException)) {
                throw $exception;
            }

            return $envelope->with(new SentToFailureTransportStamp($receivedStamp->getTransportName()));
        }
    }

    private function getNestedException(HandlerFailedException $exception): ?Throwable
    {
        $nestedExceptions = $exception->getNestedExceptions();

        if (count($nestedExceptions) === 1) {
            return $nestedExceptions[0];
        }

        return null;
    }

    private function isLastRetry(Envelope $envelope, Throwable $nestedException): bool
    {
        if ($nestedException instanceof UnrecoverableMessageHandlingException) {
            return true;
        }

        /** @var RedeliveryStamp|null $redeliveryStamp */
        $redeliveryStamp = $envelope->last(RedeliveryStamp::class);

        if ($redeliveryStamp === null) {
            return false;
        }

        return $redeliveryStamp->getRetryCount() === $this->maxRetries;
    }
}

它必须配置传输的名称和max_retries此传输的配置：

parameters:
    async_allow_failure_transport_name: 'asyncAllowFailure'
    async_allow_failure_max_retries: 5

services:
  command.bus.bypass_failure_transport_middleware:
    class: App\Infrastructure\CommandBus\Middleware\BypassFailureTransportMiddleware
    arguments:
      $transportName: '%async_allow_failure_transport_name%'
      $maxRetries: '%async_allow_failure_max_retries%'

framework:
    messenger:
        transports:
            - name: '%async_allow_failure_transport_name%'
              dsn: '...'
              retry_strategy:
                  max_retries: '%async_allow_failure_max_retries%'
                  delay: 1000
                  multiplier: 2
                  max_delay: 0

        buses:
            command.bus:
                middleware:
                  - 'command.bus.bypass_failure_transport_middleware'
于 2021-10-15T16:19:50.303 回答