.NET Core + RabbitMQ

При разработке микросервисов возникает вопрос о предоставлении информации от одного сервиса другим сервисам по мере возникновения каких-либо событий. При этом важно иметь систему без привязки отправителя к получателям. В этом случае используется паттерн Publisher-Subscriber.

На рынке существует множество продуктов для обмена сообщениями, которые поддерживают паттерн Publisher-Subscriber, например Azure Service Bus, RabbitMQ или Apache Kafka.

Недавно мною были опубликованы две NuGet библиотеки для быстрой и удобной настройки event-based communication с использованием Azure Service Bus и RabbitMQ. В этой короткой how-to статье описаны шаги для использования последней.

Исходные

Publisher — .NET Core приложение, которое выступает в роли отправителя.
Subscriber — .NET Core приложение, которое выступает в роли получателя.

How-To

1. В publisher и subscriber приложениях установите две NuGet библиотеки.

PM> Install-Package Autofac.Extensions.DependencyInjection PM> Install-Package EventBus.RabbitMQ.Standard 

2. В publisher и subscriber приложениях добавьте конфигурацию в appsettings.json.

{   "RabbitMq": {     "BrokerName": "test_broker",     "AutofacScopeName": "test_autofac",     "QueueName": "test_queue",     "RetryCount": "5",     "VirtualHost": "your_virtual_host",     "Username": "your_username",     "Password": "your_password",     "Host": "your_host",     "DispatchConsumersAsync": true   } } 

Эти настройки можно получить на CoudAMQP. Альтернативно можно использовать RabbitMQ локально (Docker image).

3. В publisher и subscriber приложениях создайте класс для события, которое будет публиковаться и, в дальнейшем, обрабатываться.

public class ItemCreatedIntegrationEvent : IntegrationEvent {     public string Title { get; set; }     public string Description { get; set; }      public ItemCreatedIntegrationEvent(string title, string description)     {         Title = title;         Description = description;     } } 

4. В subscriber приложении создайте класс для обработчика события ItemCreatedIntegrationEvent.

public class ItemCreatedIntegrationEventHandler : IIntegrationEventHandler<ItemCreatedIntegrationEvent> {     public ItemCreatedIntegrationEventHandler()     {     }      public async Task Handle(ItemCreatedIntegrationEvent @event)     {         //Handle the ItemCreatedIntegrationEvent event here.     } } 

5. В publisher и subscriber приложениях обновите Program.cs, добавив одну строчку.

public class Program {     public static void Main(string[] args)     {         CreateHostBuilder(args).Build().Run();     }      public static IHostBuilder CreateHostBuilder(string[] args) =>         Host.CreateDefaultBuilder(args)             .UseServiceProviderFactory(new AutofacServiceProviderFactory()) //Add this line.             .ConfigureWebHostDefaults(webBuilder =>             {                 webBuilder.UseStartup<Startup>();             }); } 

6. В publisher приложении обновите метод ConfigureServices в Startup.cs.

public class Startup {     public void ConfigureServices(IServiceCollection services)     {         ...          var rabbitMqOptions = Configuration.GetSection("RabbitMq").Get<RabbitMqOptions>();          services.AddRabbitMqConnection(rabbitMqOptions);         services.AddRabbitMqRegistration(rabbitMqOptions);          ...     } } 

7. В subscriber приложении создайте расширение EventBusExtension.

public static class EventBusExtension {     public static IEnumerable<IIntegrationEventHandler> GetHandlers()     {         return new List<IIntegrationEventHandler>         {             new ItemCreatedIntegrationEventHandler()         };     }      public static IApplicationBuilder SubscribeToEvents(this IApplicationBuilder app)     {         var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();          eventBus.Subscribe<ItemCreatedIntegrationEvent, ItemCreatedIntegrationEventHandler>();          return app;     } } 

8. В subscriber приложении обновите методы ConfigureServices и Configure в Startup.cs.

public class Startup {     public void ConfigureServices(IServiceCollection services)     {         ...          var rabbitMqOptions = Configuration.GetSection("RabbitMq").Get<RabbitMqOptions>();          services.AddRabbitMqConnection(rabbitMqOptions);         services.AddRabbitMqRegistration(rabbitMqOptions);         services.AddEventBusHandling(EventBusExtension.GetHandlers());          ...     }      public void Configure(IApplicationBuilder app, IWebHostEnvironment env)     {         ...          app.UseAuthorization();          ...     } } 

9. Для демонстрации и простоты можно создать контроллер в publisher приложении с методом в котором происходит публикация события.

public class ItemController : ControllerBase {     private readonly IEventBus _eventBus;      public ItemController(IEventBus eventBus)     {         _eventBus = eventBus;     }      [HttpPost]     public IActionResult Publish()     {         var message = new ItemCreatedIntegrationEvent("Item title", "Item description");          _eventBus.Publish(message);          return Ok();     } } 

10. Теперь можно опубликовать ItemCreatedIntegrationEvent. Запустите оба приложения, вызовите POST метод Publish в subscriber и обработайте событие в subscriber.

Ссылки

• EventBus.RabbitMQ.Standard
• RabbitMQ
• eShopOnContainers
• Publisher-Subscriber pattern