RabbitMQ. Partie 3. Comprendre les files d'attente et les liaisons

Queue(file d'attente) - une structure de données sur disque ou en RAM qui stocke des liens vers des messages et donne leurs copies consumers(aux consommateurs). QueueIl s'agit d'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 à l'échangeur laquelle des files d'attente doit recevoir des messages.

Table des matières

• RabbitMQ. Partie 1. Introduction. Erlang, AMQP et RPC
• RabbitMQ. Partie 2. Comprendre les échanges
• RabbitMQ. Partie 3. Comprendre les files d'attente et les liaisons
• RabbitMQ. Partie 4. Détails sur Connection et Chanel
• RabbitMQ. Partie 5. Comprendre ce que sont les messages et les cadres
• RabbitMQ. Partie 6. Publication des messages et performances de consommation
• RabbitMQ. Partie 7. Présentation des modules Fédération et Pelle
• RabbitMQ. Partie 8. RabbitMQ dans .NET
• RabbitMQ. Partie 9. Surveillance

Files d'attente temporaires

Si la file d'attente est créée avec le jeu de paramètres autoDelete, une telle file d'attente acquiert la possibilité de se supprimer automatiquement . Ces files d'attente sont généralement créées lorsque le premier client se connecte et sont supprimées lorsque tous les clients se déconnectent.

exclusive, . , /, . exclusive , autoDelete .

:

• ,
• binding churn. , / . .

durable, /. Queue.Delete.

Highly Available

HA RabbitMQ. , , .

- HA , , HA . - , .

HA .

:

• HA . HA HA RabbitMQ (2-3 )

RPC . Queue.Declare, :

RabbitMQ.Client:

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

• queue — , .
• durable — true, /
• exclusive — true,
• autoDelete — true,
• arguments — . .

arguments

• x-message-ttl(x-message-time-to-live) — . x-message-ttl, , . x-message-ttl . . . x-message-ttl, .
• x-expires — . . , , Basic.Cancel . , Basic.Get. , . , .
• x-max-length — . ,

• x-max-lenght-bytes — . ( )
• x-overflow — . : drop-head ( ) reject-publish. drop-head, . reject-publish,
• x-dead-letter-exchange — exchange, ,
• x-dead-letter-routing-key —
• x-max-priority — 255 (RabbitMQ 3.5.0 ). , . ,
• x-queue-mode — . . . , , ,
• x-queue-master-locator — ,
• x-ha-policy — HA . all, . nodes,
• x-ha-nodes — , HA

, RPC Queue.DeclareOk. ( Queue.Declare), Channel.Close OperationInterruptedException, .

Queue.Declare . , , , .

Queue.Declare , Channel.Close OperationInterruptedException, 403 .

, >= 10 , , GC , , .

Queue

RabbitMQ guest (username: guest password: guest). , guest . Queues Add a new queue. :

Add queues, .

. , consumers, / , .

Binding

RPC . Queue.Bind, :

RabbitMQ.Client:

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

• queue —
• exchange —
• routingKey —
• arguments —

, RPC Queue.BindOk.

Binding

RabbitMQ guest (username: guest password: guest). , guest . Queues my_queue. bindings:

Bind, :

Code

Dans cette section, nous décrivons la file d'attente et la liaison à l'aide du code C #, donc si nous avions besoin de développer une bibliothèque. Ce sera peut-être utile pour la perception.

public interface IQueue
    {        
        string Name { get; }

        /// <summary>
        ///       true,  queue   . 
        ///            
        ///       /. 
        ///       false,  queue     , 
        ///      /  
        /// </summary>
        bool IsDurable { get; }

        /// <summary>
        ///        true,  
        ///          
        ///       consumer-
        /// </summary>
        bool IsExclusive { get; }

        /// <summary>
        ///      . 
        ///       ,    .
        /// </summary>
        bool IsAutoDelete { get; }

        /// <summary>
        ///      
        /// </summary>
        IDictionary<string, object> Arguments { get; }
    }

public class Queue : IQueue
    {
        public Queue(
             string name, 
             bool isDurable = true, 
             bool isExclusive = false, 
             bool isAutoDelete = false, 
             IDictionary<string, object> arguments = null)
        {
            Name = name ??
                throw new ArgumentNullException(name, $"{name} must not be null");

            IsDurable = isDurable;
            IsExclusive = isExclusive;
            IsAutoDelete = isAutoDelete;
            Arguments = arguments ?? new Dictionary<string, object>();
        }

        public string Name { get; }
        public bool IsDurable { get; }
        public bool IsExclusive { get; }
        public bool IsAutoDelete { get; }
        public IDictionary<string, object> Arguments { get; }
    }

public static class QueueMode
    {       
        public const string Default = "default";
        /// <summary>
        ///      .     
        ///          ,   
        ///       
        /// </summary>
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        /// <summary>
        ///     ,    
        /// </summary>
        IExchange Exchange { get; }

        /// <summary>
        ///      
        /// </summary>
        string RoutingKey { get; }

        /// <summary>
        ///      
        /// </summary>
        IDictionary<string, object> Arguments { get; }
    }

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

        public IExchange Exchange { get; }
        public string RoutingKey { get; }
        public IDictionary<string, object> Arguments { get; }
    }