Dos herramientas que en los últimos tiempos se están convirtiendo en herramientas indispensables para los desarrolladores . Net. En este artículo veremos RabbitMQ y la librería MassTransit que son, para que la utilizamos y cómo trabajan en conjunto.
RabbitMQ
Para los que no conozcan esta herramienta, RabbitMQ es un software que nos permite manejar y negociar mensajes en un middleware. Desde nuestras aplicaciones podemos enviar mensajes a nuestro sistema u a otros que estén interesados de lo que está sucediendo y captar la información para procesar. Esto permite que nuestras aplicaciones estén desacopladas, o bien, en un arquitectura de microservicios el desacople de los mismos.
Cuando trabajamos con RabbitMQ los productores, quienes generan los eventos, pueden enviar mensajes a dos de dos maneras diferentes: Queues y Exchanges. Cuando enviamos a un mensaje una cola, éste será recibido por todos los consumidores de esa cola o Queue. Pero si deseamos enviar de manera selectiva a diferentes colas segunda metadatos usaremos Exchanges. Básicamente, cuando se recibe un mensaje, dependiendo de la configuración, se enviará a una o varias colas. Para esto existen 4 tipos de intercambio:
- Direct
- Topic
- Headers
- Fanout
Utilizaremos el tipo Fanout, este es el que usa MassTransit por defecto. Los demás puedes consultarlos de la documentación de RabbitMQ. Veamos como funciona en un gráfico.
MassTransit
Es una herramienta que nos brinda un marco para la construcción de aplicaciones distribuidas de código abierto para nuestras aplicaciones .Net. Con esta librería nos tomará menos tiempo de implementación en el trabajo con agentes como RabbitMQ creando un acoplamiento bastante flexible.
Veamos los conceptos básicos:
- Service Bus, aplicación que maneja y administra todo lo relacionado con los mensajes que viajan entre las aplicaciones.
- Mensaje, es una especie de contrato que definimos por código por medio de .Net sea por una clase o una interfaz.
- Transporte, diferentes agentes de mensajería que podemos utilizar MassTransit como RabbitMQ, En memoria, Azure Service Bus, Kafka y otros.
- Comando, es uno de los tipos de mensajes que se envían directamente a un endpoint y se declaran como una secuencia de verbos.
- Eventos, son lo que indican que algo sucedió. Los eventos se publican para uno o varios consumidores y siempre se declaran en pasado.
Usar MassTransit tiene muchas ventajas en lugar de usar los agentes propios de cada herramienta. La principal ventaja es que nos da un abstracción del agente que utilizamos, si en algún momento debemos cambiar el servicios de bus, no deberíamos cambiar nuestra implementación, solamente el proveedor.
Cuando trabajamos con mensajes deberíamos aprender varios patrones que debemos implementar como: reintentos, disyuntores, bandejas, etc. MassTransit maneja todos estos patrones haciéndonos fácil la implementación, por otro lado, también es posible manejar excepciones, transacciones distribuidas y monitoreo. Recuerdo, por más que estén implementadas, siempre es bueno saber cómo funcionan los patrones fuera de la librería.
¿Cómo lo usamos todo juntos?
Vamos a empezar a unir todo juntos. Para no tener que disponer de servidores o instalaciones en nuestro sistema operativo de las herramientas utilizaremos docker. Crearemos por un comando de docker una instancia de Rabbit para poder aprender.
docker run -d --hostname rabbitmq-server-ui --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management
Esta imagen nos permitirá usar una interfaz gráfica que estará disponible en el puerto 15672. Por otro lado el acceso al rabbitMQ será por el puerto 5672 que es el puerto predeterminado para comunicarse con él. El usuario y contraseña es guest, guest.
Lo siguiente será definir una interfaz que tenemos disponible en MassTransit que utilizaremos como contrato para los mensajes. Crearemos en nuestra aplicación un proyecto de librería compartida o shared que será consumida por el productor y el consumidor que llamaremos DemoRabbitMQ.Shared.
public interface OrderCreated
{
int Id { get; set; }
string ProductName { get; set; }
decimal Price { get; set; }
int Quantity { get; set; }
}
Recuerda que los eventos deben declararse en pasado que indica que se trata de un mensaje de evento.
Lo siguiente será crear un proyecto que llamaremos DemoRabbitMQ.Producer y le agregaremos referencia a DemoRabbitMQ.Shared. Ademas deberemos hacer referencias a los paquetes Nuget: MassTransit, MassTransit.AspNetCore y MassTransit.RabbitMQ. Dentro de de nuestro program.cs deberemos implementar el servicio para que esté disponible en la aplicación del productor.
//MassTransit
builder.Services.AddMassTransit(p =>
{
p.UsingRabbitMq();
})
Ahora crearemos OrderDto dentro de un carpeta Dtos que será utilizado por nuestro contralor:
public class OrderDto
{
public string ProductName { get; set; }
public decimal Price { get; set; }
public int Quantity { get; set; }
}
Veamos el controlador:
[ApiController]
[Route("api/[controller]")]
public class OrdersController : ControllerBase
{
private readonly IPublishEndpoint _publishEndpoint;
public OrdersController(IPublishEndpoint publishEndpoint)
{
_publishEndpoint = publishEndpoint;
}
[HttpPost]
public async Task<IActionResult> CreateOrder(OrderDto orderDto)
{
await _publishEndpoint.Publish<OrderCreated>(new
{
Id = 1,
orderDto.ProductName,
orderDto.Quantity,
orderDto.Price
});
return Ok();
}
}
Como vemos en el fragmento de código creamos un método post que se llama CreateOrder. Este método recibe nuestro Dto como parámetro para poder publicar el evento por medio del nuestro objeto publishEndpoint y utilizar OrderCreated.
Es el momento de crear al consumidor. En este caso usaremos una aplicación de consola. Como en el proyecto anterior haremos referencia al proyecto DemoRabbitMQ.Shared y debemos hacer referencia a los paquetes nuget: MassTransit y MassTransit.RabbitMQ.
Creamos una carpeta llamada Consumers y dentro crearemos una clase que implementará el consumidor con la lógica que utilizaremos para los mensajes recibidos.
class OrderCreatedConsumer : IConsumer<OrderCreated>
{
public async Task Consume(ConsumeContext<OrderCreated> context)
{
var jsonMessage = JsonConvert.SerializeObject(context.Message);
Console.WriteLine($"Mensaje: {jsonMessage}");
}
}
Agregaremos en nuestro program.cs la configuración que usara MassTransit. Este se quedará escuchando y decepcionará el evento.
using DemoRabbitMQ.Shared;
using MassTransit;
using Newtonsoft.Json;
using DemoRabbitMQ.Shared;
var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
cfg.ReceiveEndpoint("order-created-event", e =>
{
e.Consumer<OrderCreatedConsumer>();
});
});
await busControl.StartAsync(new CancellationToken());
try
{
Console.WriteLine("Press enter to exit");
await Task.Run(() => Console.ReadLine());
}
finally
{
await busControl.StopAsync();
}
class OrderCreatedConsumer : IConsumer<OrderCreated>
{
public async Task Consume(ConsumeContext<OrderCreated> context)
{
var jsonMessage = JsonConvert.SerializeObject(context.Message);
Console.WriteLine($"OrderCreated message: {jsonMessage}");
}
}
Echemos un vistazo como quedo nuestro proyecto:
Pruebas
Pongamos a prueba nuestra aplicación y servicios. Primero haremos click derecho sobre la solución y seleccionaremos las propiedades. En “Proyecto de inicio múltiple” y en acción seleccionaremos en los proyectos consumer y producer “Iniciar”. Esto hará que se ejecuten los 2 servicios:
Con F5 ejecutaremos nuestra aplicación y llamaremos con la herramienta de swagger al servicio:
El resultado que veremos en la consola que está consumiendo el evento será similar a la imagen:
Conclusiones
Hasta aquí vimos como hacer una implementación bastante simple del uso de eventos de la mano de RabbitMQ y MassTransit. Ten en cuenta que esto es solamente el puntapié inicial. Existe muchísima documentación para solucionar mucha cantidad de eventos y situaciones que podemos tener. Espero que lo hayas disfrutado.
