我正在使用.Net Core 3.1应用程序,由于某种原因,我的消息没有被使用。
services.AddMassTransit(x =>
{
x.AddConsumer<ItemAddedConsumer>();
x.AddBus(provider => Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.UseHealthCheck(provider);
cfg.Host(new Uri($"rabbitmq://{rMqSettings.host}:{rMqSettings.port}"), h =>
{
h.Username(rMqSettings.username);
h.Password(rMqSettings.password);
});
cfg.ReceiveEndpoint("items-service.ItemAdded", ep =>
{
ep.ConfigureConsumeTopology=false;
ep.Bind("ItemAdded");
ep.PrefetchCount = 15;
ep.Consumer<ItemAddedConsumer>(provider);
});
public class ItemAddedConsumer : IConsumer<ItemAdded>
{
private readonly IMediator _mediator;
public ItemAddedConsumer(IMediator mediator)
{
_mediator = mediator;
}
public async Task Consume(ConsumeContext<ItemAdded> context)
{
await _mediator.Send(new ItemAdded(context.Message.Id));
}
}
Uri uri = new Uri("exchange:ItemAdded?bind=true&queue=items-service.ItemAdded");
var endPoint = await _bus.GetSendEndpoint(uri);
await endPoint.Send(@event);
所有消息都将发送到名为Items-Service.ItemAdded_Skiped的新队列
queues
确保发送的消息和使用者使用相同的消息类型,包括命名空间as outlined in the docs。
另外,为什么发送端点地址和接收端点配置过于复杂?您可以更改消息的EntityName(通过属性或发布拓扑),并只需从消息生产者使用发布。