Реализация шины событий с помощью RabbitMQ для среды разработки или тестирования

Подсказка

Это фрагмент из электронной книги «Архитектура микрослужб .NET для контейнеризованных приложений .NET», доступной в документации .NET или в виде бесплатного скачиваемого PDF-файла, который можно прочитать в автономном режиме.

Скачать в формате PDF

Начнем с того, что если вы создаете настраиваемую шину событий на основе RabbitMQ , работающей в контейнере, так как приложение eShopOnContainers делает, оно должно использоваться только для сред разработки и тестирования. Не используйте его для рабочей среды, если только вы не создаете ее в составе служебной шины, готовой к работе, как описано в разделе "Дополнительные ресурсы" ниже. Простая настраиваемая шина событий может не иметь многих критических функций, готовых для использования в производственных условиях, которые есть у коммерческой служебной шины.

Одна из пользовательских реализаций шины событий в eShopOnContainers — это библиотека, использующая API RabbitMQ. (Существует другая реализация на основе служебной шины Azure.)

Реализация шины событий с помощью RabbitMQ позволяет микрослужбам подписываться на события, публиковать и получать их, как показано на рис. 6-21.

Рис. 6–21. Реализация шины событий с использованием RabbitMQ

RabbitMQ работает в качестве посредника между издателем сообщений и подписчиками для обработки распределения. В коде класс EventBusRabbitMQ реализует универсальный интерфейс IEventBus. Эта реализация основана на внедрении зависимостей, чтобы можно было переключить эту версию разработки и тестирования на рабочую версию.

public class EventBusRabbitMQ : IEventBus, IDisposable
{
    // Implementation using RabbitMQ API
    //...
}

Реализация RabbitMQ в примере шины событий для разработки и тестирования представляет собой шаблонный код. Он должен обрабатывать подключение к серверу RabbitMQ и предоставлять код для публикации сообщения-события в очереди. Он также должен реализовать словарь коллекций обработчиков событий интеграции для каждого типа события; Эти типы событий могут иметь разные экземпляры и разные подписки для каждой микрослужбы приемника, как показано на рис. 6-21.

Реализация простого метода публикации с помощью RabbitMQ

Следующий код представляет собой упрощенную версию реализации шины событий для RabbitMQ, чтобы продемонстрировать весь сценарий. Вы действительно не обрабатываете соединение таким образом. Чтобы увидеть полную реализацию, см. фактический код в репозитории dotnet-architecture/eShopOnContainers .

public class EventBusRabbitMQ : IEventBus, IDisposable
{
    // Member objects and other methods ...
    // ...

    public void Publish(IntegrationEvent @event)
    {
        var eventName = @event.GetType().Name;
        var factory = new ConnectionFactory() { HostName = _connectionString };
        using (var connection = factory.CreateConnection())
        using (var channel = connection.CreateModel())
        {
            channel.ExchangeDeclare(exchange: _brokerName,
                type: "direct");
            string message = JsonConvert.SerializeObject(@event);
            var body = Encoding.UTF8.GetBytes(message);
            channel.BasicPublish(exchange: _brokerName,
                routingKey: eventName,
                basicProperties: null,
                body: body);
       }
    }
}

Фактический код метода Publish в приложении eShopOnContainers улучшается с помощью политики повторных попыток Polly, которая повторяет задачу несколько раз, если контейнер RabbitMQ не готов. Этот сценарий может возникать, когда docker-compose запускает контейнеры; Например, контейнер RabbitMQ может начинаться медленнее, чем другие контейнеры.

Как упоминалось ранее, в RabbitMQ существует множество возможных конфигураций, поэтому этот код следует использовать только для сред разработки и тестирования.

Реализация кода подписки с помощью API RabbitMQ

Как и в коде публикации, следующий код является упрощением реализации шины событий для RabbitMQ. Опять же, вы, как правило, не должны изменять его, если вы не улучшаете его.

public class EventBusRabbitMQ : IEventBus, IDisposable
{
    // Member objects and other methods ...
    // ...

    public void Subscribe<T, TH>()
        where T : IntegrationEvent
        where TH : IIntegrationEventHandler<T>
    {
        var eventName = _subsManager.GetEventKey<T>();

        var containsKey = _subsManager.HasSubscriptionsForEvent(eventName);
        if (!containsKey)
        {
            if (!_persistentConnection.IsConnected)
            {
                _persistentConnection.TryConnect();
            }

            using (var channel = _persistentConnection.CreateModel())
            {
                channel.QueueBind(queue: _queueName,
                                    exchange: BROKER_NAME,
                                    routingKey: eventName);
            }
        }

        _subsManager.AddSubscription<T, TH>();
    }
}

Каждый тип события имеет связанный канал для получения событий из RabbitMQ. Затем можно иметь столько обработчиков событий на канал и тип события, сколько необходимо.

Метод Subscribe принимает объект IIntegrationEventHandler, который похож на метод обратного вызова в текущей микрослужбе, а также связанный объект IntegrationEvent. Затем код добавляет этот обработчик событий в список обработчиков событий, которые каждый тип события интеграции может иметь для каждой микрослужбы клиента. Если клиентский код еще не подписан на событие, код создает канал для типа события, чтобы он смог получать события в стиле push из RabbitMQ при публикации этого события из любой другой службы.

Как уже упоминалось выше, шина событий, реализованная в eShopOnContainers, имеет только учебное назначение, так как она обрабатывает только основные сценарии, поэтому она не готова к работе.

В рабочих сценариях проверьте дополнительные ресурсы ниже, характерные для RabbitMQ, и раздел "Реализация взаимодействия на основе событий между микрослужбами ".

Дополнительные ресурсы

Готовое к работе решение с поддержкой RabbitMQ.

• Peregrine Connect — упрощение интеграции с эффективным проектированием, развертыванием и управлением приложениями, API и рабочими процессами
  https://www.peregrineconnect.com/why-peregrine/rabbitmq-integration

• NServiceBus — полностью поддерживаемая коммерческая шина с расширенными средствами управления и мониторинга для .NET
  https://particular.net/

• EasyNetQ — клиент API .NET с открытым кодом для RabbitMQ
  https://easynetq.com/

• MassTransit — бесплатная платформа распределенных приложений с открытым исходным кодом для .NET
  https://masstransit-project.com/

• Rebus — Служебная шина .NET с открытым кодом
  https://github.com/rebus-org/Rebus

НазадДалее