PHP Symfony Messenger 怎样处理失败消息重试

Symfony Messenger 处理异步消息时，Handler 抛出异常并不等于消息立刻丢失。默认情况下，消息会按传输的重试策略再次投递；如果重试次数耗尽，又没有配置失败传输，消息可能被丢弃。比较稳妥的做法是把 retry_strategy 和 failure_transport 一起配置：临时故障自动退避重试，仍然失败的消息进入可查看、可选择重放的失败队列。

官方地址：https://symfony.com/doc/current/messenger.html

下面按我在项目里排查队列问题时更容易复用的顺序展开：先划分消息去向，再设定重试窗口，最后处理人工恢复和重复消费。

先把自动重试和失败留存分成两件事

Messenger 的异步传输负责把消息交给 Worker；Handler 处理失败后，重试策略决定“还要不要再试、多久再试”。失败传输则解决另一个问题：自动重试已经用完，消息仍然不能成功时，是否留下一个可供人工查看和处理的地方。

这两个概念经常被混在一起。只写了 retry_strategy，不代表消息会永久保存；只写了 failure_transport，也不代表临时网络抖动会以合适的间隔重试。生产配置通常至少要有 async 和 failed 两个传输。

给 async 传输配置重试和失败传输

下面是一份以 Doctrine 传输为例的配置。消息代理不同，DSN 可以换成 AMQP、Redis 或其他支持的传输，但重试策略和失败传输的职责不变。

# config/packages/messenger.yaml
framework:
    messenger:
        # 重试耗尽后，把消息送入 failed，而不是直接丢弃
        failure_transport: failed

        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    # 失败后最多再尝试 4 次；0 表示不重试
                    max_retries: 4
                    # 第一次重试前等待 1000 毫秒
                    delay: 1000
                    # 后续等待按 1s、2s、4s、8s 增长
                    multiplier: 2
                    # 把等待时间限制在 30 秒以内
                    max_delay: 30000
                    # 给延迟加入少量随机性，避免大量消息同时重试
                    jitter: 0.1

            failed: 'doctrine://default?queue_name=failed'

这里的 max_retries 是“失败后的再次尝试次数”，不是消息首次处理次数；delay、multiplier 和 max_delay 共同决定退避窗口。外部接口有明确的限流窗口时，固定的短重试往往不够，应按对方的恢复时间调整。

如果不同业务的失败影响不同，可以在具体传输上覆盖全局的 failure_transport。例如支付类消息进入高优先级失败队列，普通同步任务进入默认失败队列，避免人工处理时混在一起。

用 Worker 和失败命令处理完整生命周期

配置传输后，Worker 才会持续接收 async 中的消息。排查时可以先用更详细的输出确认 Worker 连接的是预期传输，再把失败消息从“看见”推进到“处理”。

# 持续消费 async；-vv 只用于排查时观察处理细节
php bin/console messenger:consume async -vv

# 查看失败传输中的消息，先确认异常类型和消息编号
php bin/console messenger:failed:show --max=20

# 查看编号为 20 的失败详情
php bin/console messenger:failed:show 20 -vv

# 修复外部依赖或业务数据后，强制重试指定消息
php bin/console messenger:failed:retry 20 30 --force

# 确认无法恢复且不应再次投递时，移除指定消息
php bin/console messenger:failed:remove 20

人工重试前先看失败原因。比如远端服务短暂不可用，适合修复依赖后重试；如果消息引用的业务对象已经不存在，盲目重试只会让失败队列反复增长。确认不需要保留时再执行 failed:remove，不要把“清理队列”当成修复本身。

用异常类型区分临时故障和永久故障

所有异常都按同一种策略重试，结果通常不理想。Symfony Messenger 提供了两种很有用的语义：RecoverableMessageHandlingException 表示暂时无法处理，UnrecoverableMessageHandlingException 表示确定不值得自动重试。它们让 Handler 把“故障判断”交给重试机制，而不是复制一套队列控制代码。

customerId syncToRemoteService($message->customerId);
        } catch (\RuntimeException $exception) {
            // 外部服务抖动交给 Messenger 的退避策略处理。
            throw new RecoverableMessageHandlingException(
                '远程服务暂时不可用',
                previous: $exception,
            );
        }
    }

    private function syncToRemoteService(int $customerId): void
    {
        // 省略远程客户端调用；不要在这里自行 while 重试。
    }
}

如果错误属于业务上不应重试的情况，它仍然可以按失败传输配置被保留下来，供人工检查；如果希望自定义一次重试的延迟，可以在可恢复异常中提供延迟参数。对于确实需要无限等待的临时问题要格外谨慎，否则失败消息可能长期占用 Worker。

让 Handler 能承受重复投递

重试机制只保证“再交给 Handler 一次”，不能保证 Handler 只执行一次。Worker 可能已经完成业务动作，却在确认消息前退出，传输随后再次投递同一消息。因此扣款、发邮件、扣库存这类操作必须设计成幂等。

实用做法是让消息携带稳定的业务幂等键，并在数据库或远端服务中记录已经处理的事件。这个键应该来自订单号、业务事件号等稳定标识，不能每次 dispatch 都随机生成一个新 UUID，否则同一业务事件的重复发送仍会被当成两次不同操作。

部署时保留旧消息能够反序列化

异步队列里的消息可能来自上一版代码。直接删除或重命名消息类、改变必填构造参数，都可能让新 Worker 无法反序列化旧消息。更稳妥的发布顺序是先让新版本兼容旧消息，必要时保留旧类和旧 Handler，等队列排空后再清理。

如果消息语义发生了变化，不要强行复用旧类名。新增一个版本化消息类，让旧消息继续由旧 Handler 处理，新消息走新 Handler，等迁移完成后再删除旧实现。

一份可复用的发布后检查清单

常见问题

没有配置 failure_transport 会怎样？

消息会先按重试策略再次投递；重试耗尽后，如果没有失败传输可接收它，消息可能被丢弃。需要人工追踪时，应显式配置失败传输。

失败消息应该继续用 messenger:consume failed 吗？

失败传输可以像普通传输一样消费，但很多场景更适合先查看详情、确认根因，再用 messenger:failed:retry 选择性恢复，避免把同一故障重新扩大。

为什么重试后还是出现重复业务动作？

因为重试会再次调用 Handler，且消息传输本身可能重复投递。需要用稳定幂等键、唯一约束或业务状态机保护不可重复的动作。

把 Symfony Messenger 的失败处理拆成“自动退避、失败留存、人工恢复、幂等执行”四层，排查时就能先判断消息在哪一层出了问题，再决定是调整配置、修复外部依赖还是处理具体失败消息。