RabbitMQ. Part 3. Understanding Queues and Bindings

Queue(queue) - a data structure on disk or in RAM that stores links to messages and gives their copies consumers(to consumers). QueueIt is an Erlang process with a state (where the messages themselves can be cached). 1 thousand queues can occupy about 80Mb.

Binding (binding) - a rule that tells the exchanger which of the queues should receive messages.

Table of contents

• RabbitMQ. Part 1. Introduction. Erlang, AMQP and RPC
• RabbitMQ. Part 2. Understanding Exchanges
• RabbitMQ. Part 3. Understanding Queues and Bindings
• RabbitMQ. Part 4. Details about Connection and Chanel
• RabbitMQ. Part 5. Understanding what messages and frames are
• RabbitMQ. Part 6. Message publishing and consumption performance
• RabbitMQ. Part 7. Overview of Federation and Shovel modules
• RabbitMQ. Part 8. RabbitMQ in .NET
• RabbitMQ. Part 9. Monitoring

Temporary queues

If the queue is created with the parameter set autoDelete, then such a queue acquires the ability to automatically delete itself . Such queues are usually created when the first client connects and are deleted when all clients disconnect.

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

In this section, we describe the queue and binding using C # code, so if we needed to develop a library. Perhaps it will be useful for 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; }
    }