Compartir a través de

Suscribirse a eventos

Sugerencia

Este contenido es un extracto del libro electrónico, ".NET Microservices Architecture for Containerized .NET Applications" (Arquitectura de microservicios de .NET para aplicaciones de .NET contenedorizadas), disponible en Documentación de .NET o como un PDF descargable y gratuito que se puede leer sin conexión.

Descargar PDF

Miniatura de la portada del libro electrónico 'Arquitectura de microservicios de .NET para aplicaciones .NET contenedorizadas'.

El primer paso para usar el bus de eventos es suscribir los microservicios a los eventos que quieren recibir. Esa funcionalidad debe realizarse en los microservicios receptores.

En el código simple siguiente se muestra lo que cada microservicio receptor necesita implementar al iniciar el servicio (es decir, en la Startup clase) para que se suscriba a los eventos que necesita. En este caso, el basket-api microservicio debe suscribirse a los mensajes de ProductPriceChangedIntegrationEvent y OrderStartedIntegrationEvent.

Por ejemplo, al suscribirse al ProductPriceChangedIntegrationEvent evento, que hace que el microservicio de cesta sea consciente de cualquier cambio en el precio del producto y le permite advertir al usuario sobre el cambio si ese producto está en la cesta del usuario.

var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();

eventBus.Subscribe<ProductPriceChangedIntegrationEvent,
                   ProductPriceChangedIntegrationEventHandler>();

eventBus.Subscribe<OrderStartedIntegrationEvent,
                   OrderStartedIntegrationEventHandler>();

Después de ejecutar este código, el microservicio de suscriptor escuchará a través de los canales de RabbitMQ. Cuando llega cualquier mensaje de tipo ProductPriceChangedIntegrationEvent, el código invoca al controlador de eventos que se pasa a él y procesa el evento.

Publicación de eventos a través del bus de eventos

Por último, el remitente del mensaje (microservicio de origen) publica los eventos de integración con código similar al ejemplo siguiente. (Este enfoque es un ejemplo simplificado que no tiene en cuenta la atomicidad). Implementaría código similar cada vez que se debe propagar un evento entre varios microservicios, normalmente justo después de confirmar datos o transacciones desde el microservicio de origen.

En primer lugar, el objeto de implementación del bus de eventos (basado en RabbitMQ o basado en un bus de servicio) se insertaría en el constructor del controlador, como en el código siguiente:

[Route("api/v1/[controller]")]
public class CatalogController : ControllerBase
{
    private readonly CatalogContext _context;
    private readonly IOptionsSnapshot<Settings> _settings;
    private readonly IEventBus _eventBus;

    public CatalogController(CatalogContext context,
        IOptionsSnapshot<Settings> settings,
        IEventBus eventBus)
    {
        _context = context;
        _settings = settings;
        _eventBus = eventBus;
    }
    // ...
}

Luego, lo utilizas desde los métodos de tu controlador, como en el método UpdateProduct.

[Route("items")]
[HttpPost]
public async Task<IActionResult> UpdateProduct([FromBody]CatalogItem product)
{
    var item = await _context.CatalogItems.SingleOrDefaultAsync(
        i => i.Id == product.Id);
    // ...
    if (item.Price != product.Price)
    {
        var oldPrice = item.Price;
        item.Price = product.Price;
        _context.CatalogItems.Update(item);
        var @event = new ProductPriceChangedIntegrationEvent(item.Id,
            item.Price,
            oldPrice);
        // Commit changes in original transaction
        await _context.SaveChangesAsync();
        // Publish integration event to the event bus
        // (RabbitMQ or a service bus underneath)
        _eventBus.Publish(@event);
        // ...
    }
    // ...
}

En este caso, dado que el microservicio de origen es un microservicio CRUD simple, ese código se coloca directamente en un controlador de API web.

En microservicios más avanzados, como cuando se usan enfoques de CQRS, se puede implementar en la CommandHandler clase , dentro del Handle() método .

Diseño de la atomicidad y la resistencia al publicar en el bus de eventos

Al publicar eventos de integración a través de un sistema de mensajería distribuido como el bus de eventos, tiene el problema de actualizar la base de datos original de forma atómica y de publicar un evento (es decir, se completan las dos operaciones o ninguna de ellas). Por ejemplo, en el ejemplo simplificado mostrado anteriormente, el código confirma los datos en la base de datos cuando se cambia el precio del producto y, a continuación, publica un mensaje ProductPriceChangedIntegrationEvent. Inicialmente, podría parecer esencial que estas dos operaciones se realicen de forma atómica. Sin embargo, si usa una transacción distribuida que implica la base de datos y el agente de mensajes, como lo hace en sistemas antiguos como Microsoft Message Queuing (MSMQ), este enfoque no se recomienda por las razones descritas por el teorema CAP.

Básicamente, se usan microservicios para crear sistemas escalables y de alta disponibilidad. Al simplificar un poco, el teorema CAP dice que no se puede crear una base de datos (distribuida) (o un microservicio que posee su modelo) que esté continuamente disponible, fuertemente coherente y tolerante a cualquier partición. Debería elegir dos de estas tres propiedades.

En las arquitecturas basadas en microservicios, debe elegir la disponibilidad y la tolerancia, y quitar énfasis a la coherencia fuerte. Por lo tanto, en la mayoría de las aplicaciones modernas basadas en microservicios, normalmente no desea usar transacciones distribuidas en mensajería, como sucede al implementar transacciones distribuidas basadas en el Coordinador de transacciones distribuidas (DTC) de Windows con MSMQ.

Volvamos al problema inicial y a su ejemplo. Si el servicio se bloquea después de actualizar la base de datos (en este caso, justo después de la línea de código con _context.SaveChangesAsync()), pero antes de que se publique el evento de integración, el sistema general podría ser incoherente. Este enfoque puede ser crítico para la empresa, en función de la operación empresarial específica con la que se trabaja.

Como se mencionó anteriormente en la sección de arquitectura, puede tener varios enfoques para tratar este problema:

En este escenario, el uso del patrón completo de Event Sourcing (ES) es uno de los mejores enfoques, si no es el mejor. Sin embargo, en muchos escenarios de aplicación, es posible que no pueda implementar un sistema ES completo. ES significa almacenar solo eventos de dominio en la base de datos transaccional, en lugar de almacenar los datos de estado actuales. Almacenar solo los eventos de dominio puede tener grandes ventajas, como tener el historial del sistema disponible y poder determinar el estado del sistema en cualquier momento del pasado. Sin embargo, la implementación de un sistema ES completo requiere rediseñar la mayor parte del sistema e introduce muchas otras complejidades y requisitos. Por ejemplo, le gustaría usar una base de datos hecha específicamente para el aprovisionamiento de eventos, como event Store o una base de datos orientada a documentos, como Azure Cosmos DB, MongoDB, Cassandra, CouchDB o RavenDB. Los orígenes de evento son un buen enfoque para este problema, pero no es la solución más sencilla a menos que ya esté familiarizado con los orígenes de eventos.

La opción para usar la minería de datos del registro de transacciones parece inicialmente transparente. Sin embargo, para usar este enfoque, el microservicio debe acoplarse al registro de transacciones de RDBMS, como el registro de transacciones de SQL Server. Este enfoque probablemente no es deseable. Otro inconveniente es que las actualizaciones de bajo nivel registradas en el registro de transacciones podrían no estar en el mismo nivel que los eventos de integración de alto nivel. Si es así, el proceso de ingeniería inversa de esas operaciones de registro de transacciones puede ser difícil.

Un enfoque equilibrado es una combinación de una tabla de base de datos transaccional y un patrón ES simplificado. Puede usar un estado como "listo para publicar el evento", que estableció en el evento original al confirmarlo en la tabla de eventos de integración. Después, intente publicar el evento en el bus de eventos. Si la acción publish-event se realiza correctamente, inicie otra transacción en el servicio de origen y mueva el estado de "listo para publicar el evento" a "evento ya publicado".

Si se produce un error en la acción de publicación de eventos en el bus de eventos, los datos seguirán siendo coherentes dentro del microservicio de origen, están todavía marcados como "listos para publicar el evento" y, con respecto al resto de los servicios, eventualmente serán coherentes. Siempre puede tener trabajos en segundo plano que comprueben el estado de las transacciones o los eventos de integración. Si el trabajo encuentra un evento en el estado "listo para publicar el evento", puede intentar volver a publicarlo en el bus de eventos.

Tenga en cuenta que, con este enfoque, solo se conservan los eventos de integración de cada microservicio de origen y solo los eventos que desea comunicar con otros microservicios o sistemas externos. En cambio, en un sistema ES completo, también se almacenan todos los eventos de dominio.

Por lo tanto, este enfoque equilibrado es un sistema ES simplificado. Necesita una lista de eventos de integración con su estado actual ("listo para publicar" frente a "publicado"). Pero solo tiene que implementar estos estados para los eventos de integración. Y en este enfoque, no es necesario almacenar todos los datos de dominio como eventos en la base de datos transaccional, como lo haría en un sistema ES completo.

Si ya usa una base de datos relacional, puede usar una tabla transaccional para almacenar eventos de integración. Para lograr la atomicidad en la aplicación, se usa un proceso de dos pasos basado en transacciones locales. Básicamente, tiene una tabla IntegrationEvent en la misma base de datos donde tiene las entidades de dominio. Esa tabla funciona como un seguro para lograr la atomicidad, de modo que los eventos de integración guardados se incluyan en las mismas transacciones con las que se confirman los datos de dominio.

Paso a paso, el proceso es similar al siguiente:

  1. La aplicación inicia una transacción de base de datos local.

  2. A continuación, actualiza el estado de las entidades de dominio e inserta un evento en la tabla de eventos de integración.

  3. Por último, confirma la transacción, por lo que obtendrá la atomicidad deseada y, a continuación,

  4. Publique el evento de alguna manera (próximo paso).

Al implementar los pasos para publicar los eventos, tiene estas opciones:

  • Publique el evento de integración justo después de confirmar la transacción y use otra transacción local para marcar los eventos de la tabla como publicados. A continuación, use la tabla igual que un artefacto para realizar un seguimiento de los eventos de integración en caso de problemas en los microservicios remotos y realizar acciones compensatorias basadas en los eventos de integración almacenados.

  • Usar la tabla como una especie de cola. Un subproceso de aplicación independiente o un proceso consulta la tabla de eventos de integración, publica los eventos en el bus de eventos y, a continuación, usa una transacción local para marcar los eventos como publicados.

En la figura 6-22 se muestra la arquitectura para el primero de estos enfoques.

Diagrama de atomicidad cuando se publica sin un microservicio de trabajo.

Figura 6-22. Atomicidad al publicar eventos en el bus de eventos

En el enfoque que se muestra en la figura 6-22 falta un microservicio de trabajo adicional que se encarga de comprobar y confirmar que los eventos de integración se han publicado correctamente. En caso de error, ese microservicio de trabajo de comprobación adicional puede leer los eventos de la tabla y volver a publicarlos, es decir, repetir el paso 2.

Acerca del segundo enfoque: se utiliza la tabla EventLog como una cola y siempre se emplea un microservicio de trabajo para publicar los mensajes. En ese caso, el proceso es similar al que se muestra en la figura 6-23. Esto muestra un microservicio adicional y la tabla es el único origen al publicar eventos.

Diagrama de atomicidad cuando se publica con un microservicio de trabajo.

Figura 6-23. Atomicidad al publicar eventos en el bus de eventos con un microservicio de trabajo

Por motivos de simplicidad, el ejemplo eShopOnContainers usa el primer enfoque (sin procesos adicionales ni microservicios de comprobador) más el bus de eventos. Sin embargo, el ejemplo de eShopOnContainers no controla todos los posibles casos de error. En una aplicación real implementada en la nube, debe adoptar el hecho de que surjan problemas finalmente y debe implementar esa lógica de comprobación y reenvío. El uso de la tabla como una cola puede ser más eficaz que el primer enfoque si tiene esa tabla como un único origen de eventos cuando los publica (con el trabajo) a través del bus de eventos.

Implementación de la atomicidad al publicar eventos de integración a través del bus de eventos

En el código siguiente se muestra cómo se puede crear una sola transacción que implique varios objetos DbContext: un contexto relacionado con los datos originales que se actualizan y el segundo contexto relacionado con la tabla IntegrationEventLog.

La transacción del código de ejemplo siguiente no será resistente si las conexiones a la base de datos tienen algún problema en el momento en que se ejecuta el código. Esto puede ocurrir en sistemas basados en la nube, como Azure SQL DB, que podrían mover bases de datos entre servidores. Para implementar transacciones resistentes en varios contextos, consulte la sección Implementación de conexiones SQL de Entity Framework Core resistentes más adelante en esta guía.

Para mayor claridad, en el ejemplo siguiente se muestra todo el proceso en un solo fragmento de código. Sin embargo, la implementación de eShopOnContainers se refactoriza y divide esta lógica en varias clases para que sea más fácil de mantener.

// Update Product from the Catalog microservice
//
public async Task<IActionResult> UpdateProduct([FromBody]CatalogItem productToUpdate)
{
  var catalogItem =
       await _catalogContext.CatalogItems.SingleOrDefaultAsync(i => i.Id ==
                                                               productToUpdate.Id);
  if (catalogItem == null) return NotFound();

  bool raiseProductPriceChangedEvent = false;
  IntegrationEvent priceChangedEvent = null;

  if (catalogItem.Price != productToUpdate.Price)
          raiseProductPriceChangedEvent = true;

  if (raiseProductPriceChangedEvent) // Create event if price has changed
  {
      var oldPrice = catalogItem.Price;
      priceChangedEvent = new ProductPriceChangedIntegrationEvent(catalogItem.Id,
                                                                  productToUpdate.Price,
                                                                  oldPrice);
  }
  // Update current product
  catalogItem = productToUpdate;

  // Just save the updated product if the Product's Price hasn't changed.
  if (!raiseProductPriceChangedEvent)
  {
      await _catalogContext.SaveChangesAsync();
  }
  else  // Publish to event bus only if product price changed
  {
        // Achieving atomicity between original DB and the IntegrationEventLog
        // with a local transaction
        using (var transaction = _catalogContext.Database.BeginTransaction())
        {
           _catalogContext.CatalogItems.Update(catalogItem);
           await _catalogContext.SaveChangesAsync();

           await _integrationEventLogService.SaveEventAsync(priceChangedEvent);

           transaction.Commit();
        }

      // Publish the integration event through the event bus
      _eventBus.Publish(priceChangedEvent);

      _integrationEventLogService.MarkEventAsPublishedAsync(
                                                priceChangedEvent);
  }

  return Ok();
}

Después de crear el evento de integración ProductPriceChangedIntegrationEvent, la transacción que almacena la operación de dominio original (actualizar el elemento de catálogo) también incluye la persistencia del evento en la tabla EventLog. Esto lo convierte en una sola transacción y siempre podrá comprobar si se enviaron mensajes de evento.

La tabla del registro de eventos se actualiza de forma atómica con la operación de base de datos original, mediante una transacción local en la misma base de datos. Si se produce un error en alguna de las operaciones, se produce una excepción y la transacción revierte cualquier operación completada, manteniendo así la coherencia entre las operaciones de dominio y los mensajes de evento guardados en la tabla.

Recepción de mensajes de suscripciones: controladores de eventos en microservicios receptores

Además de la lógica de suscripción de eventos, debe implementar el código interno para los manejadores de eventos de integración (como un método callback). El controlador de eventos es donde se especifica dónde se recibirán y procesarán los mensajes de evento de un tipo determinado.

Un controlador de eventos recibe primero una instancia de evento del bus de eventos. A continuación, busca el componente que se va a procesar relacionado con ese evento de integración, propagando y conservando el evento como un cambio en el estado en el microservicio receptor. Por ejemplo, si un evento ProductPriceChanged se origina en el microservicio de catálogo, se controla en el microservicio de cesta y cambia el estado de este microservicio de cesta del receptor, como se muestra en el código siguiente.

namespace Microsoft.eShopOnContainers.Services.Basket.API.IntegrationEvents.EventHandling
{
    public class ProductPriceChangedIntegrationEventHandler :
        IIntegrationEventHandler<ProductPriceChangedIntegrationEvent>
    {
        private readonly IBasketRepository _repository;

        public ProductPriceChangedIntegrationEventHandler(
            IBasketRepository repository)
        {
            _repository = repository;
        }

        public async Task Handle(ProductPriceChangedIntegrationEvent @event)
        {
            var userIds = await _repository.GetUsers();
            foreach (var id in userIds)
            {
                var basket = await _repository.GetBasket(id);
                await UpdatePriceInBasketItems(@event.ProductId, @event.NewPrice, basket);
            }
        }

        private async Task UpdatePriceInBasketItems(int productId, decimal newPrice,
            CustomerBasket basket)
        {
            var itemsToUpdate = basket?.Items?.Where(x => int.Parse(x.ProductId) ==
                productId).ToList();
            if (itemsToUpdate != null)
            {
                foreach (var item in itemsToUpdate)
                {
                    if(item.UnitPrice != newPrice)
                    {
                        var originalPrice = item.UnitPrice;
                        item.UnitPrice = newPrice;
                        item.OldUnitPrice = originalPrice;
                    }
                }
                await _repository.UpdateBasket(basket);
            }
        }
    }
}

El controlador de eventos debe comprobar si el producto existe en cualquiera de las instancias de cesta. También actualiza el precio del artículo para cada artículo de línea de cesta relacionado. Por último, crea una alerta que se mostrará al usuario sobre el cambio de precio, como se muestra en la figura 6-24.

Captura de pantalla de un explorador que muestra la notificación de cambio de precio en el carro del usuario.

Figura 6-24. Representación de un cambio del precio de un artículo en una cesta, comunicado por eventos de integración

Idempotencia en los eventos de mensajes de actualización

Un aspecto importante de los eventos de mensaje de actualización es que un error en cualquier punto de la comunicación debe hacer que se vuelva a intentar el mensaje. En caso contrario, es posible que una tarea en segundo plano intente publicar un evento que ya se ha publicado, lo que genera una condición de carrera. Asegúrese de que las actualizaciones sean idempotentes o que proporcionen suficiente información para asegurarse de que puede detectar un duplicado, descartarlo y devolver solo una respuesta.

Como se indicó anteriormente, la idempoencia significa que una operación se puede realizar varias veces sin cambiar el resultado. En un entorno de mensajería, como al comunicar eventos, un evento es idempotente si se puede entregar varias veces sin cambiar el resultado del microservicio receptor. Esto puede ser necesario debido a la naturaleza del propio evento o a la forma en que el sistema controla el evento. La idempotencia del mensaje es importante en cualquier aplicación que use mensajería, y no solo en las que implementan el patrón del bus de eventos.

Un ejemplo de una operación idempotent es una instrucción SQL que inserta datos en una tabla solo si esos datos aún no están en la tabla. No importa cuántas veces ejecute esa instrucción SQL; el resultado será el mismo: la tabla contendrá esos datos. La idempotencia de este tipo también puede ser necesaria cuando se trata de mensajes si estos podrían enviarse y, por tanto, procesarse más de una vez. Por ejemplo, si la lógica de reintento hace que un remitente envíe exactamente el mismo mensaje más de una vez, debe asegurarse de que es idempotente.

Es posible diseñar mensajes idempotentes. Por ejemplo, puede crear un evento que diga "establecer el precio del producto en $25" en lugar de "agregar $5 al precio del producto". Puede procesar de forma segura el primer mensaje cualquier número de veces y el resultado será el mismo. Esto no es cierto para el segundo mensaje. Pero incluso en el primer caso, es posible que no desee procesar el primer evento, ya que el sistema también podría haber enviado un evento de cambio de precio más reciente y esto sobrescribiría el nuevo precio.

Otro ejemplo podría ser un evento de pedido completado que se propaga a varios suscriptores. La aplicación tiene que asegurarse de que la información del pedido se actualiza en otros sistemas solo una vez, incluso si hay eventos de mensaje duplicados para el mismo evento de pedido completado.

Es conveniente tener algún tipo de identidad por evento para que pueda crear lógica que exija que cada evento se procese solo una vez por receptor.

Algunos procesamientos de mensajes son intrínsecamente idempotentes. Por ejemplo, si un sistema genera miniaturas de imagen, es posible que no importa cuántas veces se procese el mensaje sobre la miniatura generada; el resultado es que se generan las miniaturas y son las mismas cada vez. Por otro lado, operaciones como la llamada a una pasarela de pago para cobrar una tarjeta de crédito pueden no ser idempotentes en absoluto. En estos casos, debe asegurarse de que el procesamiento de un mensaje varias veces tiene el efecto esperado.

Recursos adicionales

Eliminación de duplicados en mensajes de eventos de integración

Puede asegurarse de que los eventos de mensaje se envían y procesan solo una vez por suscriptor en distintos niveles. Una manera es usar una característica de desduplicación que ofrece la infraestructura de mensajería que usa. Otro es implementar lógica personalizada en el microservicio de destino. Tener validaciones tanto en el nivel de transporte como en el nivel de aplicación es la mejor opción.

Eliminación de duplicados de eventos de mensaje al nivel de EventHandler

Una manera de asegurarse de que un evento solo se procesa una vez por cualquier receptor es implementando cierta lógica al procesar los eventos de mensaje en controladores de eventos. Por ejemplo, es el enfoque usado en la aplicación eShopOnContainers, como puede ver en el código fuente de la clase UserCheckoutAcceptedIntegrationEventHandler cuando recibe un UserCheckoutAcceptedIntegrationEvent evento de integración. (En este caso, CreateOrderCommand se envuelve con un IdentifiedCommand, usando eventMsg.RequestId como identificador, antes de enviarlo al controlador de comandos).

Desduplicación de mensajes al usar RabbitMQ

Cuando se producen errores de red intermitentes, los mensajes se pueden duplicar y el receptor de mensajes debe estar listo para controlar estos mensajes duplicados. Si es posible, los receptores deben controlar los mensajes de una manera idempotente, lo que es mejor que controlarlos de forma explícita mediante desduplicación.

Según la documentación de RabbitMQ, "si un mensaje se entrega a un consumidor y después se vuelve a poner en la cola (porque no se confirmó antes de desconectar la conexión del consumidor, por ejemplo), RabbitMQ establecerá la marca 'entregado de nuevo' cuando se vuelva a entregar (con independencia de que sea al mismo consumidor o a otro)".

Si se establece la marca "redelivered", el receptor debe tenerlo en cuenta, ya que es posible que el mensaje ya se haya procesado. Pero eso no está garantizado; es posible que el mensaje nunca haya llegado al receptor después de dejar el agente de mensajes, quizás debido a problemas de red. Por otro lado, si no se establece la marca "redelivered", se garantiza que el mensaje no se haya enviado más de una vez. Por tanto, el receptor debe desduplicar o procesar los mensajes de una manera idempotente solo si se establece la marca "entregado de nuevo" en el mensaje.

Recursos adicionales

Siguiente anterior