使用Rebus时如何正确处理已失效的Defer延迟超时消息

你提出的方案完全合理，这也是Rebus生态中处理不可取消延迟消息的主流实践，已经被大量生产项目验证过可行性。

针对你方案里的疑问先做明确答复 #

• 你不需要使用并发字典：Rebus的Saga处理内置并发控制，同一Saga实例同一时间只会有一个线程处理消息，SagaData的所有读写都在单线程上下文内完成，不需要线程安全的集合结构，你的判断完全正确。
• Cosmos DB的Mongo API适配Rebus.MongoDb组件的方案可行：Rebus的MongoDB Saga存储组件仅依赖标准MongoDB驱动API，Cosmos DB的Mongo API兼容层可以完整覆盖其功能需求，已有大量生产落地案例，不需要担心适配问题。

可优化的规范实现方案 #

你的逻辑已经可以直接上线使用，也可以做以下封装提升复用性：

1. 定义统一的超时消息接口，自带关联标识，不需要依赖传输层的消息ID：

public interface ITimeoutMessage
{
    Guid CorrelationToken { get; set; }
}

2. 封装通用的带超时管理的Saga基类，把注册、校验、清理逻辑抽为公共方法，避免每个Saga重复实现：

public abstract class TimedSagaData : SagaData
{
    // 键为超时消息类型的全限定名，值为当前有效超时的关联标识
    public Dictionary<string, Guid> ActiveTimeouts { get; set; } = new();
}

public abstract class TimedSaga<TSagaData> : Saga<TSagaData> where TSagaData : TimedSagaData
{
    /// <summary>
    /// 注册新的超时消息，自动覆盖同类型的旧超时
    /// </summary>
    protected void RegisterTimeout<TTimeout>(TimeSpan delay, TTimeout timeoutMessage) 
        where TTimeout : ITimeoutMessage
    {
        var newToken = Guid.NewGuid();
        timeoutMessage.CorrelationToken = newToken;
        Data.ActiveTimeouts[typeof(TTimeout).FullName] = newToken;
        Bus.Defer(delay, timeoutMessage);
    }

    /// <summary>
    /// 校验当前收到的超时消息是否有效
    /// </summary>
    protected bool IsTimeoutValid<TTimeout>(TTimeout timeoutMessage) 
        where TTimeout : ITimeoutMessage
    {
        var timeoutKey = typeof(TTimeout).FullName;
        return Data.ActiveTimeouts.TryGetValue(timeoutKey, out var validToken) 
               && validToken == timeoutMessage.CorrelationToken;
    }

    /// <summary>
    /// 手动清理指定类型的超时
    /// </summary>
    protected void ClearTimeout<TTimeout>()
    {
        var timeoutKey = typeof(TTimeout).FullName;
        Data.ActiveTimeouts.Remove(timeoutKey);
    }
}

3. 实际业务Saga中使用时只需要调用封装好的方法即可，示例逻辑：

// 收到事件需要调整超时时
public async Task Handle(EventUpdatedMessage message)
{
    // 其他业务逻辑
    RegisterTimeout(TimeSpan.FromDays(3), new EventExpiredTimeout{ EventId = message.EventId });
}

// 收到超时消息时
public async Task Handle(EventExpiredTimeout message)
{
    if(!IsTimeoutValid(message)) return;
    
    // 执行超时后的业务逻辑
    ClearTimeout<EventExpiredTimeout>();
}

额外注意事项 #

• 如果你使用Azure Service Bus作为传输层，注意其延迟消息的最大有效期为89天7小时，如果有超过该时长的超时需求，需要自行实现多级延迟调度，和你当前的逻辑不冲突。
• 使用Cosmos DB作为Saga存储时，建议将SagaData的Correlation ID设置为分区键，可以大幅提升并发处理性能。

内容的提问来源于stack exchange，提问作者G Mac