Il progetto a cui sto lavorando è basato su di un’architettura composta da più applicazioni che comunicano tra loro tramite scambio di messaggi. In genere, questa comunicazione avviene seguendo il pattern Publish/Subscribe, il quale prevede che un’applicazione possa comunicare in maniera asincrona con più entità interessate senza che ci sia accoppiamento tra le parti.
Le entità coinvolte sono le seguenti:
- Publisher – L’applicazione che produce il messaggio
- Message Broker – Il MOM(Message Oriented Middleware) che prende il messaggio e lo instrada verso un consumer
- Consumer – L’applicazione che consuma il messaggio.
Per il progetto viene utilizzato RabbitMQ, un messagging broker open source che supporta diversi protocolli e che offre numerose features.
RabbitMQ implementa il protocolo AMQP.0-9-1(Advanced Message Queuing Protocol), il quale prevede che i messaggi siano pubblicati verso delle entità di AMQP dette Exchange. Il ruolo degli Exchanges è quello di distribuire, secondo determinate regole chiamate bindings, i messaggi ricevuti ad altre entità chiamate Queue, a cui possono sottoscriversi uno o più Consumer.
L’algoritmo di routing con il quale possono essere instradati i messaggi verso le code dipende dal tipo di exchange. In AMQP ne sono definiti quattro, oltre quello di default, e presentano le seguenti caratteristiche:
- Default: ha la peculiarità che ogni nuova coda creata è collegata automaticamente ad esso con una routing key uguale al nome della coda.
- Direct: ideale per l’instradamento unicast. La consegna dei messaggi è basata su una routing key.
- FanOut: perfetto per l’instradamento broadcast in quanto inoltra i messaggi a tutte le code ad esso bindate ignorando la routing key.
- Topic: utilizzato per l’instradamento multicast. La consegna dei messaggi è basata su una routing key e un wildcard pattern utilizzato per bindare le code all’exchange.
- Headers: in questo caso lo scambio non è basato sulla routing key ma su attributi espressi come message header.
Sia le code che gli exchange hanno proprietà per le quali possono sopravvivere al riavvio del broker e possono autocancellarsi rispettivamente in mancanza di consumer o di code associate. Le code inoltre possono essere esclusive, cioè legate ad una sola connessione.
Invece di installare RabbitMQ in locale ho utilizzato un container Docker, sia per una questione di praticità dovuta al fatto che l’immagine dockerizzata è già pronta all’uso, sia per dare l’idea di utilizzare un broker istanziato su un’ambiente separato rispetto al publisher ed al subscriber che verranno definiti in seguito.
Per la creazione del container lanciamo il comando seguente con cui specifichiamo l’hostname, il suo nome e l’immagine di RabbitMQ che vogliamo istanziare:
docker run -d --hostname my-rabbit --name rabbit1 -p "5672:5672" -p "15672:15672" rabbitmq:3-management
Ho mappato, inoltre, la porta di default e quella della Web UI di management tra container e localhost per potervi accedere senza problemi. Ho scelto l’immagine di management proprio per la presenza di un’interfaccia web, raggiungibile all’indirizzo http://localhost:15672, con la quale è possibile interagire e monitorare le diverse entities del broker, un alternativa più intuitiva rispetto alla CLI (rabbitmqctl)
Utilizzando degli esempi messi a disposizione sul sito di RabbitMQ ho creato due console application in .NET Core. Una farà da Publisher e l’altra da Subscriber. RabbitMQ fornisce, tra gli altri, un client per il linguaggio .NET installabile mediante il gestore di pacchetti Nuget.
Analizziamo il produttore del messaggio che ho definito come Sender.
class Sender
{
static void Main(string[] args)
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "QueueDemo",
durable: false,
exclusive: false,
autoDelete: false,
arguments: null);
string message = "Demo Message";
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "",
routingKey: "QueueDemo",
basicProperties: null,
body: body);
Console.WriteLine("Sent {0}", message);
}
Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();
}
}
Creiamo una connessione verso l’EndPoint, nel nostro caso localhost, istanziando una ConnectionFactory. Eseguendo il programma in debug possiamo vedere che verso l’endpoint viene creata una connessione con protocollo AMQP con la porta di default di RabbitMQ.
Dichiariamo quindi una coda definendo la proprietà Name (QueueDemo). Creiamo il messaggio da inviare (una semplice stringa) e nel metodo BasicPublish indichiamo l’exchange di default (con la stringa vuota), la routing key (uguale al nome della coda) e inseriamo il messaggio da pubblicare nel body.
Lanciamo l’applicazione Sender :
Nella schermata di Overview della WebUI notiamo che il numero delle code è aumentato, infatti è stata creata la coda QueueDemo è c’è un messaggio in coda.
Nei dettagli specifici di QueueDemo possiamo vedere l’exchange a cui è collegata
e cliccando sul button Get Message(s) ci viene mostrato il Payload del nostro messeggio che corrisponde alla stringa creata nel Sender.
Il messaggio, quindi, è correttamente accodato ed è in attesa di essere consumato.
Analizziamo l’applicazione che farà da Subscriber che ho chiamato Receiver. Creiamo anche qui una connessione e dichiariamo la coda DemoQueue. Creiamo il consumer vero e proprio mediante la classe EventBasicConsumer e con il metodo Received andiamo a definire l’evento che verrà scatenato alla ricezione del messaggio. Per farlo assegniamo le proprietà del BasicDeliverEventArgs ea che contiene tutte le proprietà relative al messaggio consegnato dal broker. Infine col metodo BasicConsume avviamo il consumer appena definito.
class Receiver
{
static void Main(string[] args)
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.QueueDeclare(queue: "QueueDemo",
durable: false,
exclusive: false,
autoDelete: false,
arguments: null);
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine("Received {0}", message);
};
channel.BasicConsume(queue: "QueueDemo",
autoAck: true,
consumer: consumer);
Console.WriteLine("Press [enter] to exit.");
Console.ReadLine();
}
}
}
}
Eseguendo l’applicazione il messaggio viene consumato correttamente.
Con più messaggi in coda e più consumer in attesa, il load balancing viene gestito per default con uno scheduling Round Robin, per cui nessun consumer avrà priorità rispetto agli altri. Per essere sicuri che il messaggio sia stato consegnato al consumer possiamo modificare la modalità di acknowledge del messaggio. All’interno del metodo Received del consumer invochiamo il metodo BasicAck con il quale viene fatto l’acknowledge del messaggio. Tra gli argomenti di questo metodo c’è il Delivery Tag che identifica univocamente la consegna. Bisogna inoltre impostare nel BasicConsume l’autoAck a false.
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
Sulla WebUI possiamo vedere come il messaggio sia stato consumato correttamente e sia stato inviato un ack da parte del consumer.
RabbitMQ può essere un’ottima scelta per la comunicazione tra applicazioni, microservizi o comunque tutte quelle soluzioni software per cui sia necessario utilizzare un’architettura distribuita. Come middleware open source offre moltissimi vantaggi, tra cui il disaccopiamento tra le varie entità per cui è possibile farne il deploy separatamente, oppure la comunicazione asincrona che consente alle applicazioni di proseguire il flusso di esecuzione senza interruzioni.
A differenza della comunicazione HTTP mediante REST API, ad esempio, offre una maggiore affidabilità nello scambio di informazioni tra applicazioni, grazie sia a meccanismi di acknowledge e riconsegna dei messaggi sia a meccanismi di robustezza e ridondanza(cluster di nodi).
Naturalmente in questo articolo ho solo mostrato le caratteristiche di base di RabbitMQ che mi sono state utili per iniziare ad interagire con questa modalità di comunicazione.
In attesa di un articolo più avanzato, vi lascio il link al repository GitHub creato per l’occasione: https://github.com/intersect88/IntroToRabbitMQ.
Alla prossima
