Symfony信使:有可能不会抛出异常在去年重试?

Question

我们正在使用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队列发送消息时，我们希望失败的消息:

• notget tofailedtransport
• notmake exception bubble up

基本上，应该删除异常。

这是可能的吗?

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

Answer 1

我用中间件做到了这一点:

use SymfonyComponentMessengerEnvelope;
use SymfonyComponentMessengerExceptionHandlerFailedException;
use SymfonyComponentMessengerExceptionUnrecoverableMessageHandlingException;
use SymfonyComponentMessengerMiddlewareMiddlewareInterface;
use SymfonyComponentMessengerMiddlewareStackInterface;
use SymfonyComponentMessengerStampReceivedStamp;
use SymfonyComponentMessengerStampRedeliveryStamp;
use SymfonyComponentMessengerStampSentToFailureTransportStamp;
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: AppInfrastructureCommandBusMiddlewareBypassFailureTransportMiddleware
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'