.net - 如何在大众运输中记录失败的消息？

Question

我正在寻找一个很好的解决方案来记录失败的消息，在超过重试限制后，无需处理错误队列。到目前为止我发现了什么:

我可以从 InMemoryInboundMessageTracker 继承并覆盖 IsRetryLimitExceeded，但此时除了 id 之外没有关于消息本身的信息。

我可以实现 IInboundMessageInterceptor 并在 Pre/PostDispatch 中获取 IConsumeContext，但此时没有关于成功/失败的信息。

因此，作为解决方案，我可以在 PreDispatch 中获取 IConsumeContext 将其放入某种缓存中，然后在超过重试限制时将其从 IsRetryLimitExceeded 中的缓存中取出。

方法按以下顺序调用:IsRetryLimitExceeded -> PreDispatch -> PostDispatch

所以我找不到从缓存中删除成功处理的消息的好地方。

当然，我可以使用大小受限的缓存，但整个解决方案似乎很奇怪。

对此问题的任何想法将不胜感激。

Answer 1

我已经结束了这个解决方案:

class MessageInterceptor: IInboundMessageInterceptor
{
    public void PreDispatch(IConsumeContext context)
    {
        MessageTracker.Register(context);
    }

    public void PostDispatch(IConsumeContext context)
    {}
}

class MessageTracker: InMemoryInboundMessageTracker
{
    readonly Logger logger;

    static readonly ConcurrentDictionary<string, IConsumeContext> DispatchingCache = new ConcurrentDictionary<string, IConsumeContext>();

    public MessageTracker(int retryLimit, Logger logger)
        : base(retryLimit)
    {
        this.logger = logger;
    }

    public static void Register(IConsumeContext context)
    {
        DispatchingCache.GetOrAdd(context.MessageId, context);
    }

    public override void MessageWasReceivedSuccessfully(string id)
    {
        base.MessageWasReceivedSuccessfully(id);

        IConsumeContext value;
        DispatchingCache.TryRemove(id, out value);
    }

    public override bool IsRetryLimitExceeded(string id, out Exception retryException, out IEnumerable<Action> faultActions)
    {
        var result = base.IsRetryLimitExceeded(id, out retryException, out faultActions);

        IConsumeContext failed;
        if (!result || !DispatchingCache.TryRemove(id, out failed))
            return result;

        // --> log failed IConsumeContext with exception 

        return true;
    }
}

并插入这些类

        serviceBus = ServiceBusFactory.New(config =>
        {
            ...
            config.AddBusConfigurator(new PostCreateBusBuilderConfigurator(sb =>
            {
                var interceptorConfig = new InboundMessageInterceptorConfigurator(sb.InboundPipeline);
                interceptorConfig.Create(new MessageInterceptor());
            }));

            config.SetDefaultInboundMessageTrackerFactory(retryLimit => new MessageTracker(retryLimit, LogManager.GetCurrentClassLogger()));
        });