我正在微服务架构中实现应用程序。我想将某种事件存储附加到我的解决方案中。我是使用RabbitMQ进行消息传递的新手，可能我配置不正确。让我们简化我的解决方案，假设我有一个简单的通用服务，该服务必须向总线广播一些消息。要注册代理依赖项，我使用以下代码：
internal static IServiceCollection RegisterRabbitMQDependencies(this IServiceCollection services,
IConfiguration configuration)
{
var rabbitMQSettings = configuration
.GetSection(RabbitMQSettingsSectionKey)
.Get<RabbitMQSettings>();
services
.AddSingleton(serviceProvider => MassTransit.Bus.Factory.CreateUsingRabbitMq(configurator =>
{
configurator
.Host(rabbitMQSettings.HostName,
rabbitMQSettings.VirtualHostName,
hostConfigurator =>
{
hostConfigurator.Username(rabbitMQSettings.UserName);
hostConfigurator.Password(rabbitMQSettings.Password);
});
configurator.ExchangeType = ExchangeType.Fanout;
}))
.AddSingleton<IPublishEndpoint>(provider => provider.GetRequiredService<IBusControl>())
.AddSingleton<ISendEndpointProvider>(provider => provider.GetRequiredService<IBusControl>())
.AddSingleton<IBus>(provider => provider.GetRequiredService<IBusControl>())
.Configure<RabbitMQSettings>(configuration.GetSection(RabbitMQSettingsSectionKey));
return services;
}
由于我们的简化，它没有用户-它只需要向总线发布一些消息即可。来自服务的消息已正确发布，因为我可以在RabbitMq管理系统中注意到它们。问题是我的任何使用者都不响应相应的消息（可能是由于某些配置错误）。回到实现-我有一个事件存储（不同的容器）。我的事件存储区必须处理通过总线传播的所有集成事件。我已经在事件存储中配置了通信，如下所示：
public static IServiceCollection RegisterMessagingDependencies(this IServiceCollection services,
IConfiguration configuration) =>
services
.AddMassTransit(configurator =>
{
configurator.AddConsumer<EventHandler>();
})
.AddSingleton(provider => Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.Host("localhost", "/", h => {
h.Username("guest");
h.Password("guest");
});
}))
.RegisterEndpointsDependencies();
连接参数正确，我已经检查了几次。正如您所注意到的-我只有一个名为“ EventHandler”的使用者，该使用者必须处理所有IntegrationEvents。在这里，我附上EventHandler类的定义，它是父类：
public abstract class IntegrationEventHandler<TIntegrationEvent> : IIntegrationEventHandler<TIntegrationEvent>,
IConsumer<TIntegrationEvent> where TIntegrationEvent : class, IIntegrationEvent
{
public async Task Consume(ConsumeContext<TIntegrationEvent> context) =>
await HandleAsync(context.Message);
public abstract Task HandleAsync(TIntegrationEvent @event);
}
public sealed class EventHandler : IntegrationEventHandler<IIntegrationEvent>
{
private readonly IEventRepository _eventRepository;
public EventHandler(IEventRepository eventRepository)
{
_eventRepository = eventRepository;
}
public override async Task HandleAsync(IIntegrationEvent @event)
{
var readyToBeSavedEvent = new Event(@event);
await _eventRepository.CreateAsync(readyToBeSavedEvent);
await _eventRepository.SaveChangesAsync();
}
}
EventHandler基类实现了IConsumer接口，因此我认为应该以某种方式将发生在总线上的所有集成事件连接起来。但是问题是尽管发布了一些事件，但从未调用过HandleAsync方法。另外，我附加了负责发布事件的代码：
public async Task PublishAsync(params IIntegrationEvent[] events)
{
var globalPublicationTasks = events
.Select(@event =>
{
@event.PublishedAtUtc = DateTime.UtcNow;
@event.JsonContent = JsonConvert.SerializeObject(@event);
return _publishEndpoint.Publish(@event);
});
await Task.WhenAll(globalPublicationTasks);
}
我想念什么吗？我究竟做错了什么？我希望在总线上发布某些集成事件时会调用HandleAsync方法。谢谢你的帮助 ：）
您使用不AddMassTransit正确。您也不要在用户端启动总线。
public void ConfigureServices(IServiceCollection services)
{
services.AddMassTransit(x =>
{
x.AddConsumer<EventConsumer>();
x.UsingRabbitMq((context, cfg) =>
{
cfg.ReceiveEndpoint("event-listener", e =>
{
e.ConfigureConsumer<EventConsumer>(context);
});
});
});
services.AddMassTransitHostedService();
}
本文收集自互联网，转载请注明来源。
如有侵权，请联系 [email protected] 删除。
我来说两句