.net-core - MassTransit 故障处理配置
问题描述
我在我的应用程序中配置故障使用者时遇到问题。问题是消费的消息被传递到一个*_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
解决方案
要回答您的问题,它有几个部分,首先:
_error
当消费者抛出异常时,MassTransit 接收端点将消息移动到队列中。不建议在队列上创建接收端点,_error
也不应该这样做。
如果您只是想观察消费者是否发生故障,您可以创建一个单独的接收端点(例如fault-queue)并注册您的Fault<T>
消费者。MassTransit 将发布一条实施 的消息,Fault<T>
代理将通过接收端点将其路由到您的消费者。
但是,根据上面的示例,您正在发送请求并希望客户端知道是否发生了故障。为此,我建议使用请求客户端 - 它设置消息标头以将故障返回给请求发起者。它还允许发送响应。如果你不想等待响应,或者等着看故障是否发生,上面的故障观察器是你最好的选择。
您可以在文档中查看如何使用请求客户端。
推荐阅读
- spring-boot - Spring Cloud Vault:令牌(spring.cloud.vault.token)不能为空
- reactjs - prismic sm --setup:运行时没有这样的文件或目录
- python - 根据提到的逻辑将 Dataframe 从宽减少到长的有效方法
- python-3.x - 如何强制 Discord 在没有嵌入的情况下发送链接?
- windows - psql\i
在 Windows 10 上读取文件时如何解决权限被拒绝错误? - django - 使用序列化器如何在 django rest 框架中创建类别
- kotlin - 开放类和抽象类有什么区别?
- python - 在python中将每个字符相乘n次后在每个字符之间添加空格
- firebase - 从 firebase 存储下载图像并使用 node.js 云功能添加到 jszip
- python-3.x - 将 RSA 密钥导出到 stunnel.key