MassTransit RabbitMQ 将 Error Queue 上的 Consumed 消息的一半移动到 Error_Skipped Queue

Question

我正在尝试将消息从错误队列移回它源自的队列。 为此，我在错误队列上创建了一个消费者，然后将其发布到所需队列。 当我尝试这样做时，一半的已用消息会被发布，但另一半会被发送到 Error_Skipped Queue。

我尝试了很多事情都没有成功，所以我可能缺少一些简单的东西。

以下是我的代码示例：

public class ClaimsMessage
{
    public string Description { get; set; }

    public DateTime Date { get; set; }

    public bool Handled { get; set; }
}

public class ClaimsMessageErrorConsumer : IConsumer<Fault<ClaimsMessage>>
{
    public async Task Consume(ConsumeContext<Fault<ClaimsMessage>> context)
    {
        try
        {
            await context.Publish<ClaimsMessage>(context.Message.Message);

        }
        catch (Exception e)
        {
            string error = e.Message;
        }
    }
}

public static IBusControl CreateClaimsErrorConsumerBus(string endPoint)
{
    var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
    {
        var host = cfg.Host(new Uri("rabbitmq://localhost/"), h =>
        {
            h.Username("guest");
            h.Password("guest");
        });

        cfg.ReceiveEndpoint(host, endPoint, e =>
        {
            e.Consumer(() => new ClaimsMessageErrorConsumer());
        });
    });
    return busControl;
}

Answer 1

如果您将消息从错误队列移回处理队列，则不应调用Publish——这会将消息重新发送给所有订阅者。您已经知道队列名称，因此将消息直接发送回队列。您看到的是您在错误队列上创建了一个使用者，它为该消息创建了一个交换绑定。

所以，改为这样做：

sbc.ReceiveEndpoint("input_error", x =>
{
    // this prevents extra message bindings from being created
    x.BindMessageExchanges = false;

    x.Consumer<MyMover>(() => new MyMover(inputQueueAddress);
});

public class MyMover : 
    IConsumer<ClaimsMessage>
{
    public async Task Consume(ConsumeContext<ClaimsMessage> context)
    {
        try
        {
            var endpoint = await context.GetSendEndpoint(_inputQueueAddress);
            await endpoint.Send<ClaimsMessage>(context.Message);
        }
        catch (Exception e)
        {
            string error = e.Message;
        }
    }
}

要获得额外奖励，请复制原始邮件标头，以保持邮件的保真度。