RabbitMQ. Parte 3. Entendiendo Queues y Bindings

Cola (cola) — estructura de datos en disco o en memoria, que almacena enlaces a mensajes y entrega sus copias consumidores (consumidores). Cola representa un Proceso Erlang con estado (donde pueden almacenar los propios mensajes). 1.000 colas pueden ocupar alrededor de 80Mb.

Binding (vinculación) — regla que informa al intercambiador a qué cola deben ir los mensajes.

Tabla de contenido

• RabbitMQ. Parte 1. Introducción. Erlang, AMQP y RPC
• RabbitMQ. Parte 2. Entendiendo Exchanges
• RabbitMQ. Parte 3. Entendiendo Queues y Bindings
• RabbitMQ. Parte 4. Entendiendo qué son los mensajes y los tramas
• RabbitMQ. Parte 5. Rendimiento de la publicación y consumo de mensajes
• RabbitMQ. Parte 6. Resumen de los módulos Federation y Shovel
• RabbitMQ. Parte 7. Detalles sobre Connection y Channel
• RabbitMQ. Parte 8. RabbitMQ en .NET
• RabbitMQ. Parte 9. Monitoreo

Colas temporales

Si se crea una cola con el parámetro establecido autoDelete, esta cola adquiere la capacidad de eliminarse automáticamente. Estas colas generalmente se crean en el momento de la conexión del primer cliente y se eliminan cuando todos los clientes se han desconectado.

Si se crea una cola con el parámetro establecido exclusive, entonces esta cola permite que solo un consumidor se conecte y se elimina si se cierra el canal. Hasta que el canal no se cierre, el cliente puede desconectarse/conectarse, pero solo dentro de la misma conexión. Si el parámetro exclusive está establecido, entonces el parámetro autoDelete no tiene ningún efecto.

Características:

• en caso de una desconexión momentánea, perderemos mensajes que aún no han llegado al consumidor
• se puede experimentar el fenómeno binding churn. Este fenómeno ocurre cuando la cantidad de operaciones de creación/eliminación de colas y vinculaciones alcanza valores muy altos. En modo clúster, este flujo de operaciones se distribuirá por todos los nodos y generará una gran carga. Este proceso se puede optimizar controlando el número de suscripciones

Colas permanentes

Si se crea una cola con el parámetro establecido durable, entonces esta cola mantiene su estado y se restaura después del reinicio del servidor/intermediario. Esta cola existirá hasta que se invoque el comando Queue.Delete.

Colas altamente disponibles

Las colas HA requieren un entorno de clúster RabbitMQ. En modo clúster, toda la información sobre intercambiadores, colas, vinculaciones y consumidores se replicará en todos los nodos.

Cuando un mensaje se publica en alguna cola HA, se almacena en cada nodo relacionado con la cola HA. Después de que se consume el mensaje en uno de los nodos, todas las copias de este mensaje se eliminarán en los otros nodos.

Las colas HA pueden expandirse por todos los nodos en un clúster o solo en nodos individuales.

Características:

• El uso de colas HA conlleva penalizaciones en el rendimiento. Al colocar un mensaje en alguna cola HA o al consumir un mensaje de la cola HA, RabbitMQ debe coordinarse entre todos los nodos (2-3 nodos son generalmente suficientes).

Creación de Cola

La creación de una cola se realiza mediante una solicitud sincrónica RPC al servidor. La solicitud se lleva a cabo mediante el método Queue.Declare, que se invoca con los parámetros:

• nombre de la cola
• otros parámetros

Ejemplo de creación de una cola mediante RabbitMQ.Client:

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

• queue — el nombre de la cola que deseamos crear. El nombre debe ser único y no puede coincidir con el nombre del sistema de la cola.
• durable — si es true, la cola será capaz de conservar su estado y se restaurará después de reiniciar el servidor/broker.
• exclusive — si es true, la cola permitirá conectarse solo a un consumidor.
• autoDelete — si es true, la cola obtiene la capacidad de eliminarse automáticamente
• arguments — argumentos opcionales. A continuación, analizaremos en detalle.

arguments

• x-message-ttl(x-message-time-to-live) — permite establecer el tiempo de vida del mensaje en milisegundos. Si la creación de la cola se realiza con un valor establecido para el argumento x-message-ttl, esa cola excluirá automáticamente los mensajes cuya fecha de caducidad haya pasado.. Establecer el valor del argumento x-message-ttl define la edad máxima para todos los mensajes en esta cola. Crear dicha cola ayuda a prevenir la recepción de información desactualizada.. Esto se puede usar en sistemas en tiempo real. Si la cola para la que se ha establecido un intercambio para mensajes rechazados tiene un valor establecido para el argumento x-message-ttl, los mensajes rechazados en esta cola comenzarán a tener un tiempo de vida..
• x-expires — establece un valor en milisegundos después del cual la cola se elimina. La cola puede agotar su tiempo de vida solo si no tiene suscriptores. Si hay suscriptores conectados a la cola, solo podrá eliminarse automáticamente cuando todos los suscriptores invoquen Basic.Cancel o se desconecten. La vida útil de la cola solo puede finalizar si no ha habido solicitudes a ella Basic.Get. De lo contrario, el valor actual del tiempo de vida se restablece, y la cola ya no se eliminará automáticamente. También no hay garantías sobre la rapidez con la que se elimina la cola después de que haya caducado.
• x-max-length — define el número máximo de mensajes en la cola. Si la cantidad de mensajes en la cola comienza a superar la cantidad máxima, se eliminarán los más antiguos

• x-max-length-bytes — establece el tamaño total máximo permitido de la carga útil de los mensajes en la cola. Cuando se supera el valor establecido (se produce un desbordamiento de la cola al publicar un nuevo mensaje), se comenzarán a eliminar los mensajes más antiguos
• x-overflow — este argumento se utiliza para configurar el comportamiento como resultado del desbordamiento de la cola. Hay dos valores disponibles: drop-head (valor predeterminado) y reject-publish. Si se selecciona drop-head, los mensajes más antiguos se eliminarán. Si se elige reject-publish, la aceptación de mensajes se suspenderá
• x-dead-letter-exchange — define el exchange al que se envían los mensajes rechazados que no se vuelven a poner en cola
• x-dead-letter-routing-key — define una clave de enrutamiento opcional para los mensajes rechazados
• x-max-priority — permite la clasificación por prioridades en la cola con un valor de prioridad máxima de 255 (RabbitMQ versiones 3.5.0 y superiores). El número indica la prioridad máxima que admitirá la cola. Si el argumento no está establecido, la cola no admitirá la prioridad de los mensajes
• x-queue-mode — permite poner la cola en modo perezoso. En este modo, se almacenarán en disco la mayor cantidad posible de mensajes. El uso de la memoria RAM será mínimo. Si no se establece, la cola almacenará los mensajes en memoria para entregar los mensajes lo más rápido posible
• x-queue-master-locator — si tenemos un clúster, se puede definir la cola maestra
• x-ha-policy — se utiliza al crear colas HA y define cómo se distribuirá el mensaje entre los nodos. Si se establece el valor todo, el mensaje se guardará en todos los nodos. Si se establece el valor nodes, el mensaje se guardará en nodos específicos del clúster
• x-ha-nodes — define los nodos que estarán relacionados con una determinada cola HA

Si se crea una cola posiblemente, el servidor enviará al cliente un RPC una respuesta Queue.DeclareOk. Si la creación de la cola no es posible (se produjo un error en la solicitud Queue.Declare), entonces el canal se cerrará por el servidor mediante el comando Channel.Close y el cliente recibirá una excepción OperationInterruptedException, que contendrá un código de error y su descripción.

Llamada repetida Queue.Declare con parámetros similares devolverá información útil sobre esta cola. Por ejemplo, el número total de mensajes en espera en esta cola y el número total de consumidores suscritos a ella.

mountsnoop.py Queue.Declare con las credenciales de un usuario que no tiene los derechos necesarios cerrará el canal mediante el comando Channel.Close y el cliente recibirá una excepción OperationInterruptedException, que contendrá un código de error 403 y su descripción.

Después de que la cola haya estado inactiva durante >= 10 segundos, ella entra en modo de suspensión, llamando a GC en la cola, lo que resulta en una reducción significativa de la memoria necesaria para esta cola.

Creación de Queue a través de la interfaz gráfica

Accedemos al panel de administración RabbitMQ con el usuario invitado (nombre de usuario: invitado y contraseña: invitado). Tenga en cuenta que el usuario invitado solo puede conectarse desde el host local. Ahora vamos a la pestaña Colas y hacemos clic en Agregar una nueva cola. Llenamos las propiedades:

Una vez que introduzcamos todos los datos necesarios y hagamos clic en Agregar colas, la cola aparecerá en la lista general.

Hacer clic en el nombre de la cola mostrará su información detallada. Aquí se puede configurar la vinculación entre el intercambio y la cola, ver la lista consumidores, publicar/recibir mensajes, eliminar la cola y ver las estadísticas.

Creación de Binding

La creación de la vinculación se realiza mediante la llamada sincrónica RPC al servidor. La solicitud se lleva a cabo mediante el método Queue.Bind, que se invoca con los parámetros:

• nombre de la cola
• nombre del punto de intercambio
• otros parámetros

Ejemplo de creación de una vinculación mediante RabbitMQ.Client:

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

• queue — nombre de la cola
• exchange — nombre del intercambiador
• routingKey — clave de enrutamiento
• arguments — argumentos opcionales

Si la creación de la vinculación posiblemente, el servidor enviará al cliente un RPC una respuesta Queue.BindOk.

Creación de Binding a través de la interfaz gráfica

Accedemos al panel de administración RabbitMQ con el usuario invitado (nombre de usuario: invitado y contraseña: invitado). Tenga en cuenta que el usuario invitado solo puede conectarse desde el host local. Ahora vamos a la pestaña Colas y hacemos clic en la cola my_queue. Llenamos los campos de la sección enlaces:

Una vez que introduzcamos todos los datos necesarios y hagamos clic en Vincular, la vinculación se mostrará en la lista general:

Código

En esta sección describiremos la cola y la vinculación con código en C#, como si tuviéramos que desarrollar una biblioteca. Esto podría ser útil para la comprensión.

interfaz pública IQueue
    {        
        cadena Nombre { obtener; }

        // 
        //     Si se establece en verdadero, la cola será persistente. 
        //     Se almacenará en el disco y podrá 
        //     sobrevivir a un reinicio del servidor/broker. 
        //     Si el valor es falso, la cola será temporal y se eliminará, 
        //     cuando el servidor/broker se reinicie
        // 
        bool EsDuradera { obtener; }

        // 
        //     Si el valor es verdadero, 
        //     tal cola permitirá la conexión 
        //     solo a un consumidor
        // 
        bool EsExclusiva { obtener; }

        // 
        //     Eliminación automática. 
        //     La cola se eliminará cuando todos los clientes se desconecten.
        // 
        bool EsAutoEliminar { obtener; }

        // 
        //     Argumentos opcionales
        // 
        IDictionary Argumentos { obtener; }
    }

clase pública Queue : IQueue
    {
        public Queue(
             cadena nombre, 
             bool esDuradera = verdadero, 
             bool esExclusiva = falso, 
             bool esAutoEliminar = falso, 
             IDictionary argumentos = nulo)
        {
            Nombre = nombre ??
                lanzar nueva ArgumentNullException(nombre, $"{nombre} no debe ser nulo");

            EsDuradera = esDuradera;
            EsExclusiva = esExclusiva;
            EsAutoEliminar = esAutoEliminar;
            Argumentos = argumentos ?? nueva Dictionary();
        }

        public cadena Nombre { obtener; }
        public bool EsDuradera { obtener; }
        public bool EsExclusiva { obtener; }
        public bool EsAutoEliminar { obtener; }
        public IDictionary Argumentos { obtener; }
    }

clase pública estática QueueMode
    {       
        public const string Predeterminado = "default";
        // 
        //     Modo perezoso. El modo perezoso hará que se guarden 
        //     la mayor cantidad posible de mensajes en el disco para reducir 
        //     el uso de memoria
        // 
        public const string Perezoso = "lazy";
    }

interfaz pública IBinding
    {
        // 
        //     Intercambiador que se conectará con el enlace
        // 
        IExchange Intercambio { obtener; }

        // 
        //     Clave de enrutamiento
        // 
        cadena ClaveDeEnrutamiento { obtener; }

        // 
        //     Argumentos opcionales
        // 
        IDictionary Argumentos { obtener; }
    }

clase pública Binding : IBinding
    {
        public Binding(
             IExchange intercambio, 
             cadena claveDeEnrutamiento, 
             IDictionary argumentos)
        {
            Intercambio = intercambio;
            ClaveDeEnrutamiento = claveDeEnrutamiento;
            Argumentos = argumentos;
        }

        public IExchange Intercambio { obtener; }
        public cadena ClaveDeEnrutamiento { obtener; }
        public IDictionary Argumentos { obtener; }
    }

Fuente: habr.com