RabbitMQ. Parte 3. Approfondiamo Queues e Bindings

Coda (coda) — struttura dati su disco o in memoria che conserva riferimenti ai messaggi e restituisce copie di essi consumatori (ai consumatori). Coda è rappresentato da un processo Erlang con stato (dove possono essere memorizzati anche i messaggi). 1.000 code possono occupare circa 80Mb.

Binding (legame) — è una regola che informa l'exchange a quale coda devono arrivare i messaggi.

Indice

• RabbitMQ. Parte 1. Introduzione. Erlang, AMQP e RPC
• RabbitMQ. Parte 2. Approfondiamo gli Exchanges
• RabbitMQ. Parte 3. Approfondiamo Queues e Bindings
• RabbitMQ. Parte 4. Analizziamo cosa sono i messaggi e i frame
• RabbitMQ. Parte 5. Performance di pubblicazione e consumo dei messaggi
• RabbitMQ. Parte 6. Panoramica sui moduli Federation e Shovel
• RabbitMQ. Parte 7. Dettagli su Connection e Channel
• RabbitMQ. Parte 8. RabbitMQ in .NET
• RabbitMQ. Parte 9. Monitoraggio

Code temporanee

Se la creazione di una coda avviene con il parametro impostato autoDelete, allora tale coda acquisisce la capacità di eliminarsi automaticamente. Tali code vengono solitamente create al momento della connessione del primo client e rimosse quando tutti i client si disconnettono.

Se la creazione di una coda avviene con il parametro impostato esclusiva, allora tale coda consente di collegare solo un utente e viene rimossa se il canale si chiude. Finché il canale rimane attivo, il cliente può disconnettersi/ricollegarsi, ma solo all'interno della stessa connessione. Se il parametro esclusiva è impostato, il parametro autoDelete non ha alcun effetto.

Caratteristiche:

• In caso di breve interruzione della connessione, perderemo i messaggi che non sono ancora stati consegnati all'utente
• si può osservare il fenomeno binding churn. Il fenomeno si verifica quando il numero di operazioni di creazione/rimozione di code e collegamenti raggiunge valori molto elevati. In modalità cluster, questo flusso di operazioni si diffonderà su tutti i nodi, creando un alto carico. Questo processo può essere ottimizzato controllando il numero di sottoscrizioni.

Code persistenti

Se la creazione di una coda avviene con il parametro impostato durable, allora tale coda mantiene il proprio stato e si ripristina dopo il riavvio del server/broker. Questa coda esisterà finché non verrà invocato il comando Queue.Delete.

Code ad alta disponibilità

Le code HA richiedono un ambiente cluster RabbitMQ. In modalità cluster, tutte le informazioni su exchange, code, collegamenti e consumatori saranno replicate su tutti i nodi.

Quando un messaggio viene pubblicato in una coda HA, viene memorizzato su ciascun nodo associato a quella coda HA. Dopo che il messaggio è stato consumato su uno dei nodi, tutte le copie di quel messaggio verranno eliminate sugli altri nodi.

Le code HA possono estendersi a tutti i nodi in un certo cluster o solo a nodi specifici.

Caratteristiche:

• l'uso delle code HA comporta un impatto sulle prestazioni. Quando un messaggio viene inserito in una coda HA o consumato da essa, RabbitMQ deve coordinare l'operazione su tutti i nodi (2-3 nodi sono generalmente sufficienti).

Creazione della coda

La creazione della coda avviene tramite una richiesta RPC al server. La richiesta viene effettuata utilizzando il metodo Queue.Declare, chiamato con i seguenti parametri:

• nome della coda
• altri parametri

Esempio di creazione di una coda utilizzando RabbitMQ.Client:

// ...
channel.QueueDeclare(
    queue: "my_queue",
    durable: false,
    exclusive: false,
    autoDelete: false,
    arguments: null
);
// ...

• queue — è il nome della coda che desideriamo creare. Il nome deve essere unico e non può coincidere con il nome di sistema della coda.
• durable — se true, la coda sarà in grado di mantenere il proprio stato e verrà ripristinata dopo il riavvio del server/broker.
• esclusiva — se true, la coda permetterà di connettersi solo a un consumatore
• autoDelete — se true, la coda acquisisce la capacità di eliminarsi automaticamente
• argomenti — argomenti opzionali. Approfondiamo di seguito.

argomenti

• x-message-ttl(x-message-time-to-live) — consente di impostare il tempo di scadenza del messaggio in millisecondi. Se la coda viene creata con un valore impostato per l'argomento x-message-ttl, tale coda escluderà automaticamente i messaggi scaduti. Impostare il valore dell'argomento x-message-ttl definisce l'età massima per tutti i messaggi in questa coda. Creare tale coda permette di evitare di ricevere informazioni obsolete. Questo può essere utilizzato in sistemi in tempo reale. Se per la coda per la quale è impostato lo scambio per i messaggi rifiutati si imposta il valore dell'argomento x-message-ttl, i messaggi rifiutati in questa coda inizieranno ad avere un tempo di vita.
• x-expires — imposta il valore in millisecondi dopo il quale avviene la cancellazione della coda. La coda può scadere solo se non ha alcun abbonato. Se ci sono abbonati connessi alla coda, essa potrà essere rimossa automaticamente solo quando tutti gli abbonati disconnettono Basic.Cancel o si disconnettono. La durata della vita della coda può terminare solo se non ci sono state richieste Basic.Get. Altrimenti, il valore attuale della durata viene azzerato e la coda non verrà più eliminata automaticamente. Inoltre non ci sono garanzie su quanto rapidamente avvenga la cancellazione della coda dopo la scadenza della sua durata.
• x-max-length — imposta il numero massimo di messaggi nella coda. Se il numero di messaggi nella coda supera il massimo, i messaggi più vecchi iniziano ad essere rimossi

• x-max-lenght-bytes — imposta la dimensione massima consentita per il carico utile dei messaggi nella coda. Se viene superato il valore impostato (si è verificato un overflow della coda durante la pubblicazione di un messaggio), i messaggi più vecchi iniziano ad essere rimossi
• x-overflow — questo argomento è utilizzato per configurare il comportamento in caso di overflow della coda. Sono disponibili due valori: drop-head (valore predefinito) e reject-publish. Se si sceglie drop-head, i messaggi più vecchi verranno eliminati. Se si sceglie reject-publish, l'accettazione dei messaggi sarà sospesa
• x-dead-letter-exchange — specifica l'exchange in cui vengono inviati i messaggi rifiutati che non vengono reinseriti in coda
• x-dead-letter-routing-key — definisce una chiave di instradamento opzionale per i messaggi rifiutati
• x-max-priority — consente l'ordinamento per priorità nella coda, con un valore massimo di priorità di 255 (RabbitMQ versioni 3.5.0 e successive). Questo numero indica la priorità massima che supporterà la coda. Se l'argomento non è impostato, la coda non supporterà la priorità dei messaggi
• x-queue-mode — consente di impostare la coda in modalità pigra. In questa modalità il maggior numero possibile di messaggi verrà memorizzato su disco. L'uso della memoria sarà minimo. Se non è impostato, la coda memorizzerà i messaggi in memoria per consegnarli il più rapidamente possibile
• x-queue-master-locator — se abbiamo un cluster, è possibile impostare una coda master
• x-ha-policy — utilizzato nella creazione di code HA e definisce come il messaggio sarà distribuito tra i nodi. Se è impostato il valore all, il messaggio sarà memorizzato su tutti i nodi. Se è impostato il valore nodes, il messaggio sarà memorizzato su nodi specifici del cluster
• x-ha-nodes — specifica i nodi a cui appartiene una certa coda HA

Se la creazione della coda è possibile, il server invierà al cliente una risposta sincrona richiesta RPC risposta Queue.DeclareOk. Se la creazione della coda non è possibile (si è verificato un errore nella richiesta Queue.Declare), allora il canale verrà chiuso dal server tramite il comando Channel.Close e il cliente riceverà un'eccezione OperationInterruptedException, che conterrà il codice di errore e la sua descrizione.

Una nuova chiamata Queue.Declare con parametri simili restituirà informazioni utili su questa coda. Ad esempio, il numero totale di messaggi in attesa in questa coda e il numero totale di consumatori ad essa iscritti.

La chiamata Queue.Declare con le credenziali dell'utente a cui non sono stati concessi i privilegi necessari chiuderà il canale utilizzando il comando Channel.Close e il cliente riceverà un'eccezione OperationInterruptedException, che conterrà il codice di errore 403 e la sua descrizione.

Dopo che la coda è inattiva per almeno 10 secondi, essa entra in modalità sleep, attivando la GC nella coda, il che porta a una significativa riduzione della memoria necessaria per questa coda.

Creazione della Queue tramite l'interfaccia grafica

Accediamo al pannello di amministrazione RabbitMQ con l'utente guest (username: guest e password: guest). Si prega di notare che l'utente guest può collegarsi solo dall'host locale. Procediamo ora alla scheda Queues e facciamo clic su Add a new queue. Compiliamo le proprietà:

Non appena inseriamo tutti i dati necessari e facciamo clic su Add queues, la coda apparirà nell'elenco generale.

Cliccando sul nome della coda si mostreranno i dettagli. Qui è possibile configurare il binding tra l'exchange e la coda, visualizzare l'elenco consumatori, pubblicare/ricevere messaggi, eliminare la coda e visualizzare le statistiche.

Creazione Binding

La creazione del binding avviene tramite il comando sincrono richiesta RPC al server. La richiesta viene effettuata utilizzando il metodo Queue.Bind, chiamato con i seguenti parametri:

• nome della coda
• nome dell'exchange
• altri parametri

Esempio di creazione binding utilizzando RabbitMQ.Client:

//...
channel.QueueBind(
    queue: queueName,
    exchange: "my_exchange",
    routingKey: "my_key",
    arguments: null
);
//...

• queue — nome della coda
• exchange — nome dell'exchange
• routingKey — chiave di routing
• argomenti — argomenti opzionali

Se la creazione del binding è possibile, il server invierà al cliente una risposta sincrona richiesta RPC risposta Queue.BindOk.

Creazione Binding tramite interfaccia grafica

Accediamo al pannello di amministrazione RabbitMQ con l'utente guest (username: guest e password: guest). Si prega di notare che l'utente guest può collegarsi solo dall'host locale. Procediamo ora alla scheda Queues e clicchiamo sulla coda my_queue. Compilando i campi della sezione bindings:

Appena inseriamo tutti i dati necessari e premiamo su Bind, il binding verrà mostrato nell'elenco generale:

Code

In questa sezione descriveremo la coda e il binding usando il codice C#, come se dovessimo sviluppare una libreria. Questo potrebbe essere utile per la comprensione.

public interface IQueue
    {        
        string Name { get; }

        // 

        //     Se impostato su true, la queue sarà persistente. 
        //     Sarà memorizzata su disco e in grado di 
        //     sopravvivere al riavvio del server/broker. 
        //     Se false, la queue sarà temporanea e verrà eliminata, 
        //     quando il server/broker verrà riavviato
        // 
        bool IsDurable { get; }

        // 

        //     Se true, 
        //     questa coda consentirà la connessione 
        //     solo a un consumer
        // 
        bool IsExclusive { get; }

        // 

        //     Eliminazione automatica. 
        //     La coda sarà rimossa quando tutti i client si disconnettono.
        // 
        bool IsAutoDelete { get; }

        // 

        //     Argomenti opzionali
        // 
        IDictionary Arguments { get; }
    }

public class Queue : IQueue
    {
        public Queue(
             string name, 
             bool isDurable = true, 
             bool isExclusive = false, 
             bool isAutoDelete = false, 
             IDictionary arguments = null)
        {
            Name = name ??
                throw new ArgumentNullException(name, $"{name} non può essere nullo");

            IsDurable = isDurable;
            IsExclusive = isExclusive;
            IsAutoDelete = isAutoDelete;
            Arguments = arguments ?? new Dictionary();
        }

        public string Name { get; }
        public bool IsDurable { get; }
        public bool IsExclusive { get; }
        public bool IsAutoDelete { get; }
        public IDictionary Arguments { get; }
    }

public static class QueueMode
    {       
        public const string Default = "default";
        // 

        //     Modalità pigra. La modalità pigra costringe a salvare 
        //     il maggior numero possibile di messaggi su disco per ridurre 
        //     l'uso della memoria
        // 
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // 

        //     Exchange che sarà collegato tramite binding
        // 
        IExchange Exchange { get; }

        // 

        //     Chiave di routing
        // 
        string RoutingKey { get; }

        // 

        //     Argomenti opzionali
        // 
        IDictionary Arguments { get; }
    }

public class Binding : IBinding
    {
        public Binding(
             IExchange exchange, 
             string routingKey, 
             IDictionary arguments)
        {
            Exchange = exchange;
            RoutingKey = routingKey;
            Arguments = arguments;
        }

        public IExchange Exchange { get; }
        public string RoutingKey { get; }
        public IDictionary Arguments { get; }
    }

Fonte: habr.com