Symfony Messenger的重试与限速：利用自定义中间件控制消息处理速率 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Symfony Messenger 的重试与限速：利用自定义中间件控制消息处理速率

大家好，今天我们来深入探讨 Symfony Messenger 的一个重要方面：消息的重试与限速。在实际应用中，消息队列的处理并非总是顺利的，可能会遇到各种问题，比如外部服务不可用、数据库连接超时等等。我们需要一种机制来处理这些瞬时错误，并防止系统因为大量的失败消息而过载。Symfony Messenger 提供了强大的中间件机制，允许我们自定义消息处理流程，从而实现重试和限速的功能。

一、重试机制：应对瞬时错误

重试机制是处理瞬时错误的有效手段。当消息处理失败时，我们不立即放弃，而是尝试重新处理。Symfony Messenger 内置了重试传输 (Retryable Transport) 的概念，但它主要适用于传输层面的错误，例如连接失败。对于应用层面的错误，我们需要自定义中间件来实现更精细的控制。

1.1 为什么需要自定义重试中间件？

Symfony Messenger 的 FailedMessageProcessingMiddleware 可以将失败的消息移动到失败队列（Failed Transport）。但是，我们可能希望在将消息移动到失败队列之前，进行多次重试。另外，我们可能希望根据不同的异常类型，采取不同的重试策略。例如，对于网络超时，我们可能希望立即重试，而对于业务逻辑错误，我们可能希望直接放弃。

1.2 实现自定义重试中间件

下面是一个自定义重试中间件的例子：

<?php

namespace AppMessengerMiddleware;

use PsrLogLoggerInterface;
use SymfonyComponentMessengerEnvelope;
use SymfonyComponentMessengerMiddlewareMiddlewareInterface;
use SymfonyComponentMessengerMiddlewareStackInterface;
use SymfonyComponentMessengerStampRedeliveryStamp;
use Throwable;

class RetryMessageMiddleware implements MiddlewareInterface
{
    private LoggerInterface $logger;
    private int $maxRetries;
    private int $delay; // 延迟时间，单位毫秒

    public function __construct(LoggerInterface $logger, int $maxRetries = 3, int $delay = 1000)
    {
        $this->logger = $logger;
        $this->maxRetries = $maxRetries;
        $this->delay = $delay;
    }

    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        try {
            return $stack->next()->handle($envelope, $stack);
        } catch (Throwable $exception) {
            $redeliveryStamp = $envelope->last(RedeliveryStamp::class);
            $retryCount = $redeliveryStamp ? $redeliveryStamp->getRetryCount() : 0;

            if ($retryCount < $this->maxRetries) {
                $this->logger->warning('Message failed, retrying...', [
                    'message' => $envelope->getMessage(),
                    'exception' => $exception,
                    'retryCount' => $retryCount + 1,
                    'maxRetries' => $this->maxRetries,
                ]);

                // 添加 RedeliveryStamp，并设置延迟
                $envelope = $envelope->with(new RedeliveryStamp($retryCount + 1, $this->delay));

                // 重新发布消息到消息队列
                // 注意：这里没有直接重新处理消息，而是将其重新放回队列，
                // 让消费者下次再消费。  可以考虑直接递归调用 handle 方法，
                // 但需要注意避免无限循环。
                return $envelope;
            } else {
                $this->logger->error('Message failed after max retries.', [
                    'message' => $envelope->getMessage(),
                    'exception' => $exception,
                    'retryCount' => $retryCount + 1,
                    'maxRetries' => $this->maxRetries,
                ]);

                // 抛出异常，交给 FailedMessageProcessingMiddleware 处理
                throw $exception;
            }
        }
    }
}

代码解释：

__construct(): 构造函数，接收 LoggerInterface 实例、最大重试次数和延迟时间作为参数。
handle(): 中间件的核心方法。
- 首先尝试调用下一个中间件处理消息。
- 如果发生异常，获取 RedeliveryStamp，如果不存在则创建。
- 检查重试次数是否小于最大重试次数。
- 如果可以重试，记录日志，添加 RedeliveryStamp，并重新发布消息到消息队列。
- 如果达到最大重试次数，记录错误日志，并抛出原始异常，让 FailedMessageProcessingMiddleware 处理。

1.3 配置中间件

在 messenger.yaml 中配置中间件：

framework:
    messenger:
        buses:
            command.bus:
                middleware:
                    - AppMessengerMiddlewareRetryMessageMiddleware
                    - doctrine.orm.messenger.middleware.doctrine_transaction
                    - validation
                    - AppMessengerMiddlewareRateLimitMiddleware # RateLimitMiddleware 示例，稍后介绍

1.4 RedeliveryStamp 的作用

RedeliveryStamp 是 Symfony Messenger 提供的一个 Stamp，用于携带消息的重试信息。它包含以下属性：

retryCount: 当前重试次数。
delay: 延迟时间，单位毫秒。

通过 RedeliveryStamp，我们可以控制消息的重试次数和重试间隔。

1.5 改进：基于异常类型进行重试

我们可以根据不同的异常类型，决定是否进行重试。例如：

    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        try {
            return $stack->next()->handle($envelope, $stack);
        } catch (Throwable $exception) {
            $redeliveryStamp = $envelope->last(RedeliveryStamp::class);
            $retryCount = $redeliveryStamp ? $redeliveryStamp->getRetryCount() : 0;

            // 根据异常类型判断是否重试
            if ($exception instanceof SomeSpecificException && $retryCount < $this->maxRetries) {
                // ... 重试逻辑
            } elseif ($retryCount < $this->maxRetries) {
                // ... 通用重试逻辑
            } else {
                // ... 失败处理
            }
        }
    }

二、限速机制：防止系统过载

限速机制用于控制消息的处理速率，防止系统因为大量的消息处理而过载。这在处理来自外部系统的大量请求时尤其重要。

2.1 为什么需要限速中间件？

在高并发场景下，大量的消息涌入可能会导致系统资源耗尽，服务响应变慢甚至崩溃。限速可以保证系统在可承受的范围内运行，避免雪崩效应。

2.2 实现自定义限速中间件

下面是一个自定义限速中间件的例子，使用了漏桶算法：

<?php

namespace AppMessengerMiddleware;

use PsrLogLoggerInterface;
use SymfonyComponentMessengerEnvelope;
use SymfonyComponentMessengerMiddlewareMiddlewareInterface;
use SymfonyComponentMessengerMiddlewareStackInterface;
use SymfonyComponentCacheAdapterAdapterInterface;
use SymfonyComponentLockLockFactory;

class RateLimitMiddleware implements MiddlewareInterface
{
    private LoggerInterface $logger;
    private AdapterInterface $cache;
    private int $capacity; // 桶的容量
    private int $rate; // 消息处理速率 (每秒处理多少个消息)
    private LockFactory $lockFactory;
    private string $cacheKeyPrefix = 'rate_limit_';

    public function __construct(LoggerInterface $logger, AdapterInterface $cache, int $capacity = 10, int $rate = 5, LockFactory $lockFactory)
    {
        $this->logger = $logger;
        $this->cache = $cache;
        $this->capacity = $capacity;
        $this->rate = $rate;
        $this->lockFactory = $lockFactory;
    }

    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        $messageName = get_class($envelope->getMessage());
        $cacheKey = $this->cacheKeyPrefix . md5($messageName);

        $lock = $this->lockFactory->createLock($cacheKey); // 使用锁防止并发问题

        if ($lock->acquire()) {
            $currentLevel = $this->cache->getItem($cacheKey)->get() ?: 0; // 获取当前水位

            // 计算时间间隔，确保速率符合要求
            $timeSinceLastUpdate = (microtime(true) - ($this->cache->getItem($cacheKey . '_timestamp')->get() ?: 0));
            $currentLevel = max(0, $currentLevel - ($timeSinceLastUpdate * $this->rate)); // 漏水

            if ($currentLevel < $this->capacity) {
                $currentLevel++; // 加水
                $this->cache->getItem($cacheKey)->set($currentLevel);
                $this->cache->getItem($cacheKey . '_timestamp')->set(microtime(true));
                $this->cache->saveMultiple([
                  $this->cache->getItem($cacheKey),
                  $this->cache->getItem($cacheKey . '_timestamp')
                ]);

                $lock->release();
                return $stack->next()->handle($envelope, $stack);
            } else {
                $lock->release();
                $this->logger->warning('Rate limit exceeded for message.', [
                    'message' => $envelope->getMessage(),
                    'capacity' => $this->capacity,
                    'rate' => $this->rate,
                ]);

                // 抛出异常，让 FailedMessageProcessingMiddleware 处理，或者延迟处理
                // 这里可以选择不同的策略，例如抛出异常，或者将消息放回队列等待下次处理
                throw new Exception('Rate limit exceeded');
                //return $envelope; // 如果选择延迟处理，可以直接返回 $envelope，让消息重新入队
            }
        } else {
            // 无法获取锁，说明有其他进程正在处理，等待下次重试
            $this->logger->warning('Could not acquire lock for rate limiting, retrying later.', [
                'message' => $envelope->getMessage(),
            ]);
            throw new Exception('Could not acquire lock for rate limiting');
        }
    }
}

代码解释：

__construct(): 构造函数，接收 LoggerInterface 实例、Cache 实例、桶的容量和消息处理速率作为参数。
handle(): 中间件的核心方法。
- 使用 Cache 组件来存储当前桶的水位。
- 使用 LockFactory 来防止并发访问 Cache 造成数据不一致的问题。
- 计算自上次更新以来的时间，模拟漏桶的 "漏水" 过程。
- 如果桶未满，则将消息加入桶中，并更新缓存。
- 如果桶已满，则记录警告日志，并抛出异常，或者将消息放回队列。

2.3 配置中间件

在 services.yaml 中定义 LockFactory：

services:
    AppMessengerMiddlewareRateLimitMiddleware:
        arguments:
            $lockFactory: '@lock.factory'

    lock.factory:
        class: SymfonyComponentLockLockFactory
        arguments:
            - '@lock.store'

    lock.store:
        class: SymfonyComponentLockStoreFlockStore
        arguments:
            - '%kernel.cache_dir%/locks'

确保 messenger.yaml 中已经配置了 RateLimitMiddleware (如之前的例子所示)。

2.4 漏桶算法的原理

漏桶算法的核心思想是将请求像水滴一样注入到固定容量的桶中。桶以恒定的速率漏水，如果请求到达的速度超过了漏水的速度，桶就会溢出，溢出的请求会被丢弃或延迟处理。

桶的容量 (Capacity): 决定了可以容纳的最大请求数量。
漏水速率 (Rate): 决定了每秒可以处理的请求数量。

2.5 算法选择：漏桶 vs 令牌桶

除了漏桶算法，还有令牌桶算法也是一种常见的限流算法。

特性	漏桶算法	令牌桶算法
原理	请求像水滴注入桶，桶以恒定速率漏水，溢出丢弃	以恒定速率向桶中放入令牌，请求需要获取令牌才能通过
平滑流量	更好，流量更平稳	允许一定程度的突发流量
适用场景	严格限制平均速率的场景	允许一定突发流量的场景

选择哪种算法取决于你的具体需求。如果需要严格控制平均速率，防止出现突发流量，那么漏桶算法更适合。如果允许一定程度的突发流量，并且对平均速率要求不高，那么令牌桶算法更适合。

三、结合重试和限速：更健壮的消息处理

我们可以将重试和限速结合起来，构建更健壮的消息处理流程。例如，如果消息因为限速而被拒绝，我们可以将消息重新放回队列，等待下次处理。同时，我们也可以对失败的消息进行重试，提高消息处理的成功率。

3.1 中间件的顺序

中间件的顺序非常重要。一般来说，应该先进行限速，再进行重试。这样可以避免因为大量的失败消息重试而导致系统过载。

framework:
    messenger:
        buses:
            command.bus:
                middleware:
                    - AppMessengerMiddlewareRateLimitMiddleware
                    - AppMessengerMiddlewareRetryMessageMiddleware
                    - doctrine.orm.messenger.middleware.doctrine_transaction
                    - validation

3.2 异常处理策略

在限速中间件中，如果消息被拒绝，可以选择抛出异常，或者将消息放回队列。如果选择抛出异常，那么重试中间件可以捕获这个异常，并进行重试。

四、其他注意事项

监控和告警： 监控消息队列的长度、消息处理的成功率和失败率，并设置告警阈值，以便及时发现和解决问题。
消息过期时间： 为消息设置过期时间，防止消息在队列中积压过久。
死信队列： 将处理失败的消息移动到死信队列，以便后续分析和处理。
幂等性： 确保消息处理的幂等性，避免因为消息重复处理而导致数据错误。

五、总结：灵活控制消息处理，保障系统稳定

这篇文章介绍了如何使用 Symfony Messenger 的中间件机制来实现消息的重试和限速。自定义中间件可以帮助我们更灵活地控制消息处理流程，应对各种异常情况，并防止系统过载。通过合理地配置中间件和选择合适的算法，我们可以构建更健壮、更可靠的消息处理系统。

Symfony Messenger 的重试与限速：利用自定义中间件控制消息处理速率

发表回复 取消回复

发表回复取消回复