🥇RabbitMQ. Partie 3. Comprendre les Queues et Bindings

Queue (file d'attente) — une structure de données sur disque ou en mémoire qui stocke des références aux messages et en restitue des copies. consumers (pour les consommateurs). Queue est un Processus Erlang avec un état (où les messages eux-mêmes peuvent être mis en cache). 1 000 files d'attente peuvent occuper environ 80 Mo.

Binding (liaison) — une règle qui indique au courtier dans quelle file d'attente les messages doivent être envoyés.

Table des matières

RabbitMQ. Partie 1. Introduction. Erlang, AMQP et RPC
RabbitMQ. Partie 2. Comprenons les Exchanges
RabbitMQ. Partie 3. Comprendre les Queues et les Bindings
RabbitMQ. Partie 4. Comprendre ce que sont les messages et les trames
RabbitMQ. Partie 5. Performance de publication et de consommation des messages
RabbitMQ. Partie 6. Aperçu des modules Federation et Shovel
RabbitMQ. Partie 7. Détails sur la connexion et le canal
RabbitMQ. Partie 8. RabbitMQ sous .NET
RabbitMQ. Partie 9. Surveillance

Files d'attente temporaires

Si la création de la file d'attente se fait avec le paramètre autoDelete, alors cette file d'attente acquiert la capacité de se supprimer automatiquement. Ces files d'attente sont généralement créées lors de la connexion du premier client et supprimées lorsque tous les clients se déconnectent.

Si la création de la file d'attente se fait avec le paramètre exclusive, alors cette file d'attente n'autorise qu'un seul consommateur à se connecter et est supprimée si le canal se ferme. Tant que le canal n'est pas fermé, le client peut se déconnecter/se reconnecter, mais uniquement dans le cadre de la même connexion. Si le paramètre exclusive est fixé, alors le paramètre autoDelete n'a aucun effet.

Caractéristiques :

en cas de coupure temporaire de la connexion, nous perdrons des messages qui n'ont pas encore atteint le consommateur.
on peut observer le phénomène binding churn. Ce phénomène se produit lorsque le nombre d'opérations de création/suppression de files d'attente et de liaisons atteint des valeurs très élevées. En mode cluster, ce flux d'opérations se répartira sur tous les nœuds et créera une charge importante. Ce processus peut être optimisé en contrôlant le nombre d'abonnements.

Files d'attente permanentes

Si la création de la file d'attente se fait avec le paramètre durable, alors cette file d'attente conservent leur état et se restaurent après le redémarrage du serveur/broker. Cette file d'attente existera tant qu'une commande Queue.Delete.

Les files d'attente hautement disponibles

Les files d'attente HA nécessitent un environnement cluster RabbitMQ. En mode cluster, toutes les informations concernant les courtiers, les files d'attente, les liaisons et les consommateurs seront copiées sur tous les nœuds.

Lorsque un message est publié dans une file d'attente HA, il est stocké sur chaque nœud associé à la file d'attente HA. Une fois que le message a été consommé sur l'un des nœuds, toutes les copies de ce message seront supprimées sur les autres nœuds.

Les files d'attente HA peuvent s'étendre sur tous les nœuds d'un certain cluster ou seulement sur des nœuds individuels.

Caractéristiques :

L'utilisation des files d'attente HA entraîne des pénalités en termes de performances. Lorsque vous placez un message dans une file d'attente HA ou lorsque vous consommez un message d'une file d'attente HA, RabbitMQ doit effectuer une coordination entre tous les nœuds (2-3 nœuds sont généralement suffisants).

Création d'une file d'attente

La création d'une file d'attente se fait par le biais d'une synchronisation. RPC synchronisation au serveur. La demande est réalisée par la méthode Queue.Declare, appelée avec les paramètres :

nom de la file d'attente
autres paramètres

Exemple de création d'une file d'attente par le biais de RabbitMQ.Client:

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

queue — le nom de la file d'attente que nous souhaitons créer. Le nom doit être unique et ne peut pas coïncider avec le nom système de la file d'attente.
durable — si true, la file d'attente sera conserve son état et se restaure après le redémarrage du serveur/broker.
exclusive — si true, la file d'attente n'autorisera qu'un seul consommateur à se connecter.
autoDelete — si true, la file d'attente acquiert la capacité de se supprimer automatiquement
arguments — arguments optionnels. Nous allons les examiner en détail ci-dessous.

arguments

x-message-ttl(x-message-time-to-live) — permet de définir la durée de vie d'un message en millisecondes. Si la création de la file d'attente se fait avec une valeur d'argument définie, x-message-ttlcette file d'attente va exclure automatiquement les messages qui ont expiré.En définissant la valeur de l'argument, x-message-ttl vous spécifiez l'âge maximum pour tous les messages de cette file d'attente. Création de cette file d'attente permet d'éviter de recevoir des informations obsolètes.Cela peut être utilisé dans des systèmes en temps réel. Si une file d'attente pour laquelle un échange a été spécifié pour les messages rejetés reçoit une valeur d'argument, x-message-ttlles messages rejetés dans cette file d'attente commenceront à avoir une durée de vie..
x-expires — définit une valeur en millisecondes après laquelle la file d'attente est supprimée. La file d'attente peut expirer uniquement si elle n'a aucun abonné. Si la file d'attente a des abonnés connectés, elle ne pourra être automatiquement supprimée que lorsque tous les abonnés auront invoqué Basic.Cancel ou se déconnecteront. La durée de vie de la file d'attente peut prendre fin uniquement si aucune demande n'a été faite à celle-ci, Basic.Get.Sinon, la valeur actuelle de la durée de vie est réinitialisée et la file d'attente ne sera plus automatiquement supprimée. De plus, il n'y a aucune garantie sur la rapidité avec laquelle la file d'attente est supprimée après l'expiration de sa durée de vie..
x-max-length — définit le nombre maximum de messages dans la file d'attente. Si le nombre de messages dans la file d'attente commence à dépasser le nombre maximum, les plus anciens messages commencent à être supprimés.

x-max-length-bytes — définit la taille totale maximale autorisée des charges utiles des messages dans la file d'attente. En cas de dépassement de la valeur fixée (une saturation de la file d'attente se produit lors de la publication d'un message), les plus anciens messages commenceront à être supprimés.
x-overflow — cet argument est utilisé pour configurer le comportement en cas de saturation de la file d'attente. Deux valeurs sont disponibles : drop-head (valeur par défaut) et reject-publish. Si vous choisissez drop-head, alors les plus anciens messages seront supprimés. Si vous choisissez reject-publish, la réception des messages sera suspendue.
x-dead-letter-exchange — définit l'échange vers lequel sont envoyés les messages rejetés, qui ne sont pas retentés dans la file d'attente.
x-dead-letter-routing-key — définit une clé de routage optionnelle pour les messages rejetés.
x-max-priority — permet le tri par priorités dans la file d'attente avec une valeur maximale de priorité de 255 (RabbitMQ version 3.5.0 et supérieure). Le nombre indique la priorité maximale que la file d'attente supportera. Si l'argument n'est pas défini, la file d'attente ne supportera pas les priorités des messages.
x-queue-mode — permet de transférer la file d'attente en mode paresseux. En ce mode, un maximum de messages sera stocké sur le disque. L'utilisation de la mémoire vive sera minimale. Si ce mode n'est pas défini, la file d'attente stockera les messages en mémoire pour livrer les messages le plus rapidement possible.
x-queue-master-locator — si nous avons un cluster, nous pouvons désigner la file d'attente maître.
x-ha-policy — utilisé lors de la création de files d'attente HA et définit comment le message sera distribué sur les nœuds. Si la valeur est définie sur tout, alors le message sera conservé sur tous les nœuds. Si la valeur est définie sur nodes, alors le message sera conservé sur certains nœuds du cluster.
x-ha-nodes — définit les nœuds auxquels une certaine file d'attente sera associée. HA

Si la création de la file d'attente peut-être, alors le serveur enverra au client un RPC réponse Queue.DeclareOk. Si la création de la file d'attente est impossible (une erreur de demande est survenue Queue.Declare), alors le canal sera fermé par le serveur avec la commande Channel.Close et le client recevra une exception OperationInterruptedException, qui contiendra un code d'erreur et sa description.

Un nouvel appel Queue.Declare avec des paramètres similaires renverra des informations utiles sur cette file d'attente. Par exemple, le nombre total de messages en attente dans cette file d'attente et le nombre total de consommateurs abonnés.

de system-nspawn Queue.Declare avec les identifiants d'un utilisateur qui n'a pas les privilèges nécessaires fermera le canal en utilisant la commande Channel.Close et le client recevra une exception OperationInterruptedException, qui contiendra le code d'erreur 403 et sa description.

Après qu'une file d'attente ait été inactive pendant >= 10 secondes, elle entre en mode veille, déclenchant le GC dans la file, ce qui entraîne une réduction significative de la mémoire nécessaire pour cette file.

Création d'une file d'attente via l'interface graphique

Accédez au panneau d'administration RabbitMQ sous l'utilisateur guest (nom d'utilisateur : guest et mot de passe : guest). Notez que l'utilisateur guest ne peut se connecter qu'à partir de l'hôte local. Passons maintenant à l'onglet Files d'attente et cliquez sur Ajouter une nouvelle file d'attente. Remplissez les propriétés :

Une fois que nous aurons saisi toutes les données nécessaires et cliqué sur Ajouter des files d'attente, la file apparaîtra dans la liste générale.

Cliquez sur le nom de la file d'attente pour afficher ses informations détaillées. Ici, vous pouvez configurer l'association entre l'échange et la file, voir la liste consumers, publier/récupérer des messages, supprimer la file et consulter les statistiques.

Création d'un Binding

La création d'un binding se fait à l'aide de l'appel synchrone RPC synchronisation au serveur. La demande est réalisée par la méthode Queue.Bind, appelée avec les paramètres :

nom de la file d'attente
nom de la touche d'échange
autres paramètres

Exemple de création d'un binding à l'aide de RabbitMQ.Client:

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

queue — nom de la file d'attente
l'échange — nom de l'échange
routingKey — clé de routage
arguments — arguments optionnels

Si la création du binding peut-être, alors le serveur enverra au client un RPC réponse Queue.BindOk.

Création d'un Binding via l'interface graphique

Accédez au panneau d'administration RabbitMQ sous l'utilisateur guest (nom d'utilisateur : guest et mot de passe : guest). Notez que l'utilisateur guest ne peut se connecter qu'à partir de l'hôte local. Passons maintenant à l'onglet Files d'attente et nous cliquons sur la file d'attente my_queue. Remplissons les champs de la section bindings:

Une fois que nous aurons saisi toutes les données nécessaires et cliqué sur Bind, le binding apparaîtra dans la liste générale :

Code

Dans cette section, nous allons décrire la file d'attente et le binding en code C#, comme si nous devions développer une bibliothèque. Cela pourrait être utile pour la compréhension.

public interface IQueue
    {        
        string Name { get; }

        // <summary>
        //     Si défini sur true, la file sera persistante. 
        //     Elle sera stockée sur le disque et pourra 
        //     survivre au redémarrage du serveur du courtier. 
        //     Si la valeur est false, la file est temporaire et sera supprimée, 
        //     lorsque le serveur du courtier sera redémarré
        // <summary>
        bool IsDurable { get; }

        // <summary>
        //     Si la valeur est true, 
        //     cette file ne permettra à un seul client de se connecter
        // <summary>
        bool IsExclusive { get; }

        // <summary>
        //     Suppression automatique. 
        //     La file sera supprimée lorsque tous les clients se déconnecteront.
        // <summary>
        bool IsAutoDelete { get; }

        // <summary>
        //     Arguments optionnels
        // <summary>
        IDictionary<string, object> 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} ne doit pas être nul");

            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";
        // <summary>
        //     Mode paresseux. Le mode paresseux forcera à conserver 
        //     autant de messages que possible sur le disque pour réduire 
        //     l'utilisation de la mémoire vive
        // </summary>
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // <summary>
        //     Échange qui sera associé par liaison
        // </summary>
        IExchange Exchange { get; }

        // <summary>
        //     Clé de routage
        // </summary>
        string RoutingKey { get; }

        // <summary>
        //     Arguments optionnels
        // </summary>
        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; }
    }

Source : habr.com

RabbitMQ. Partie 3. Comprendre les Queues et les Bindings