MassTransit故障处理配置

Question

我在应用程序中配置故障消费者时遇到问题。问题是，消耗的消息被传递到*_error_skipped队列，并且它并没有完全消失。

下面是一个非常简单的例子。客户端应用程序接收到失败消息，该消息从test_error队列中消失，但仍存在于test_error_skipped队列中。

服务项目

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using GreenPipes;
using MassTransit;
using MassTransit.Util;
namespace MassTransitTest.Service
{
public class RequestModel
{
public DateTime RequestTime { get; set; }
}
class MassTransitService : IDisposable
{
private readonly IBusControl _busControl;
public MassTransitService()
{
_busControl = Bus.Factory.CreateUsingRabbitMq(configure =>
{
var host = configure.Host(new Uri("rabbitmq://localhost/mt_test"), h =>
{
h.Username("guest");
h.Password("guest");
});
configure.ReceiveEndpoint(host, "test", c =>
{
c.UseRetry(r => r.None());
c.Consumer<RequestConsumer>();
});
});
TaskUtil.Await(_busControl.StartAsync());
Console.WriteLine("bus started");
}
public void Dispose()
{
_busControl?.StopAsync().Wait();
}
}
class RequestConsumer : IConsumer<RequestModel>
{
public Task Consume(ConsumeContext<RequestModel> context)
{
Console.WriteLine($"request with message id {context.MessageId} received: {context.Message.RequestTime}");
throw new NotImplementedException();
}
}
}

客户项目

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using GreenPipes;
using MassTransit;
using MassTransit.Util;
using MassTransitTest.Service;
namespace MassTransitTest.Client
{
class MassTransitClient
{
private readonly IBusControl _busControl;
public MassTransitClient()
{
_busControl = Bus.Factory.CreateUsingRabbitMq(configure =>
{
var host = configure.Host(new Uri("rabbitmq://localhost/mt_test"), h =>
{
h.Username("guest");
h.Password("guest");
}); enter code here
configure.ReceiveEndpoint(host, "test_error", c =>
{
c.Consumer<ErrorConsumer>();
});
});
TaskUtil.Await(_busControl.StartAsync());
Console.WriteLine("bus started");
}
public async Task Send()
{
Console.WriteLine("sending request");
await (await _busControl.GetSendEndpoint(new Uri("rabbitmq://localhost/mt_test/test"))).Send(new RequestModel()
{
RequestTime = DateTime.Now
});
Console.WriteLine("request sent");
}
}
class ErrorConsumer : IConsumer<Fault<RequestModel>>
{
public Task Consume(ConsumeContext<Fault<RequestModel>> context)
{
Console.WriteLine($"request with message id {context.Message.FaultedMessageId} failed. requested time: {context.Message.Message.RequestTime}");
return Task.CompletedTask;
}
}
}

我使用的是.net core 2.1.1和MassTransit 5.1.3

Answer 1

要回答您的问题，有几个部分，首先：

当使用者引发异常时，MassTransit接收端点会将消息移动到_error队列。不建议也不应该在_error队列上创建接收端点。

如果您只是想观察使用者是否发生了故障，可以创建一个单独的接收端点(如故障队列(并注册Fault<T>使用者。MassTransit将发布一条实现Fault<T>的消息，代理将通过接收端点将其路由到您的消费者。

但是，根据上面的示例，您发送了一个请求，并期望客户端知道是否发生了故障。为此，我建议使用请求客户端，它设置消息头以将错误返回给请求发起者。它还允许发送响应。如果您不想等待响应，或者不想等待故障是否发生，那么上述故障观测器是您的最佳选择。

您可以在文档中看到如何使用请求客户端。