Rebus 重试策略

Question

我对以下 Rebus 重试政策有疑问：

Configure.With(...)
    .Options(b => b.SimpleRetryStrategy(maxDeliveryAttempts: 2))
    .(...)

https://github.com/rebus-org/Rebus/wiki/Automatic-retries-and-error-handling#customizing-retries

1 可以同时用于发布者（入队消息）和订阅者（​​出队消息）吗？

2 我有一个无法将消息出列的订阅者。因此消息被发送到错误队列。

以下是将消息放入错误队列时的错误。但是我看不到重试的日志。

[ERR] Rebus.Retry.PoisonQueues.PoisonQueueErrorHandler (Thread #9): Moving messa
ge with ID "<unknown>" to error queue "poison"
Rebus.Exceptions.RebusApplicationException: Received message with empty or absen
t 'rbs2-msg-id' header! All messages must be supplied with an ID . If no ID is p
resent, the message cannot be tracked between delivery attempts, and other stuff
 would also be much harder to do - therefore, it is a requirement that messages
be supplied with an ID.

是否可以为每次重试定义和存储自定义日志记录，而不是在 IErrorHandler 中？

3 每次重试之间的默认等待时间是多久？

4 是否可以为每次重试定义自定义等待时间（不在 IErrorHandler 中）？如果是这样，此扫描仪是否支持 Polly？如下：

Random jitterer = new Random(); 
Policy
  .Handle<HttpResponseException>() // etc
  .WaitAndRetry(5,    // exponential back-off plus some jitter
      retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt))  
                    + TimeSpan.FromMilliseconds(jitterer.Next(0, 100)) 
  );

https://docs.microsoft.com/en-us/dotnet/standard/microservices-architecture/implement-resilient-applications/implement-http-call-retries-exponential-backoff-polly

更新

如何测试重试策略？

以下是我根据以下代码尝试的：

public class StringMessageHandler : IHandleMessages<String>
{   
    public async Task Handle(String message) 
    {
          //retry logic with Polly here
    }   
}

我向字符串主题发送了一个字符串类型的无效消息，但是根本没有调用 Handle(String message)。

Answer 1

Rebus 的重试机制仅在接收消息时相关。它的工作原理是创建一个“队列事务”(*)，然后如果消息处理失败，消息会回滚到队列中。

此后几乎立即，将再次接收并尝试处理该消息。这意味着交付尝试之间没有延迟。

对于每个失败的传递，该消息的 ID 都会增加一个计数器。这就是 Rebus 工作需要消息 ID 的原因，这也解释了为什么没有 ID 的消息会立即移动到死信队列。

由于传递尝试的不连贯性（每个消息 ID 仅存储一个计数器），因此没有好地方可以挂钩像 Polly 这样的重试库。

如果您想对消息处理进行轮询，我建议您使用 Polly 策略执行单独的操作——这样，您可以轻松地使用不同的策略来处理失败的 Web 请求、网络驱动器上的文件传输失败等。我愿意我自己也这么多。

为避免在 Polly 重试过程中无法正确关闭总线实例，您可以将 Rebus 的内部 CancellationToken 传递给 Polly 执行，如下所示：

public class PollyMessageHandler : IHandleMessages<SomeMessage>
{
    static readonly IAsyncPolicy RetryPolicy = Policy
        .Handle<Exception>()
        .WaitAndRetryForeverAsync(_ => TimeSpan.FromSeconds(10));

    readonly IMessageContext _messageContext;

    public PollyMessageHandler(IMessageContext messageContext)
    {
        _messageContext = messageContext;
    }

    public async Task Handle(SomeMessage message) 
    {
        var cancellationToken = _messageContext.GetCancellationToken();

        await RetryPolicy.ExecuteAsync(DoStuffThatCanFail, cancellationToken);
    }

    async Task DoStuffThatCanFail(CancellationToken cancellationToken)
    {
        // do your risky stuff in here
    }
}

---

(*) 实际的事务类型取决于传输支持的内容。

对于 MSMQ，它是一个 MessageQueueTransaction 对象，上面有 Commit() 和 Rollback() 方法。

对于 RabbitMQ、Azure 服务总线和其他协议，它是基于租约的协议，其中消息在一段时间内变得不可见，然后如果在这段时间内消息被确认，则消息被删除。否则——如果消息或 NACK，或者如果租约到期——消息会再次弹出，并且可以再次被其他消费者接收。

使用 SQL 传输，它只是一个数据库事务。