Masstransit 消耗故障而不将其从 _error 队列中取出

Question

您好，我是公共交通和 RabbitMQ 的菜鸟。我正在尝试实现一个错误消费者，它只会从错误队列中读取，而不会实际消费并将消息从错误队列中取出。

我设置了一个消费者和一个错误消费者。

cfg.ReceiveEndpoint(host, nameof(MassMailJobCommand), e =>
                    {
                        e.Consumer<MassMailJobCommandConsumer>(context);

                        e.Consumer<FaultConsumer<MassMailJobCommand>>(context);
                    });

这是我的 FaultConsumer 类

public class FaultConsumer<T> : IConsumer<Fault<T>>
{
    private INotificationsClient _notificationsClient;
    private NotificationsOptions _notificationsOptions;
    private ILogger _logger;

    public FaultConsumer(ILogger<T> logger,
                         IOptionsSnapshot<NotificationsOptions> notificationsOptions,
                         INotificationsClient notificationClient
                                       )
    {
        _notificationsClient = notificationClient;
        _notificationsOptions = notificationsOptions.Value;
        _logger = logger;
    }

    public async Task Consume(ConsumeContext<Fault<T>> context)
    {
        string exceptions = "";
        foreach (var ex in context.Message.Exceptions)
        {
            exceptions += $"{ex.StackTrace.ToString()};" +
                "\n=================>>>>>>>>====================>>>>>>>>==========================\n" +
                "======================        *END OF STACKTRACE*         =========================\n" +
                "===================>>>>>>>>====================>>>>>>>>==========================\n";
        }

        string message = $"Rabbit MQ Error:\n*Destination Address:* {context.DestinationAddress}\n*Original Message:* {context.Message.Message}\n*Exceptions:* {exceptions}";
        await _notificationsClient.Slack(message);
    }
}

它将它发送到一个松弛通道。它工作得很好，除了我不希望消息离开错误队列。

非常感谢。

Answer 1

失败的消息总是移动到_error 队列。这就是传输方式的一部分。

为Fault<T> 创建消费者完全是一个单独的问题，它会在消息出现故障时使用由 MassTransit 发布的故障事件（并移至上述错误队列）。

故障事件和将消息移动到错误队列都作为 MassTransit 的一部分发生。

另外，作为旁注，我建议将您的故障消费者放在单独的端点上（例如fault-consumers），这样故障消息就不会与其他要处理的消息竞争。接收从队列中读取的端点并根据类型分派消息，因此一堆错误可能会出现在新命令之前。