RabbitMQ. Bahin 3. Pagsabot sa mga Pila ug Pagbugkos

Queue (pila) - usa ka istruktura sa datos sa disk o sa RAM nga nagtipig mga link sa mga mensahe ug naghatag mga kopya niini consumers (sa mga konsumidor). Queue mao ang usa ka Erlang proseso stateful (diin ang mga mensahe mismo mahimong ma-cache). Ang 1 ka libo nga pila mahimong moabot ug mga 80Mb.

Binding (pagbugkos) - usa ka lagda nga nagsulti sa tigbaylo kung asa nga mga mensahe sa pila ang kinahanglan mahulog.

Mga sulud sa sulud

• RabbitMQ. Bahin 1. Pasiuna. Erlang, AMQP ug RPC
• RabbitMQ. Bahin 2. Pagsabot sa Pagbayloay
• RabbitMQ. Bahin 3. Pagsabot sa mga Pila ug Pagbugkos
• RabbitMQ. Bahin 4. Pag-atubang sa unsa ang mga mensahe ug mga bayanan
• RabbitMQ. Bahin 5: Pagmantala ug Pagkonsumo sa Pagganap sa Mensahe
• RabbitMQ. Bahin 6. Kinatibuk-ang Pagtan-aw sa Federation ug Mga Module sa Pala
• RabbitMQ. Bahin 7. Mga Detalye bahin sa Koneksyon ug Chanel
• RabbitMQ. Bahin 8. RabbitMQ sa .NET
• RabbitMQ. Bahin 9. Pagmonitor

Temporaryo nga Pila

Kung ang paghimo sa pila mahitabo sa set parameter autoDelete, unya ang ingon nga pila nakakuha sa abilidad awtomatik nga tangtangon ang imong kaugalingon. Ang ingon nga mga pila sagad gihimo kung ang una nga kliyente nagkonektar ug mapapas kung ang tanan nga mga kliyente nadiskonekta.

Kung ang paghimo sa pila mahitabo sa set parameter exclusive, unya ang ingon nga pila nagtugot lamang sa usa ka user sa pagkonektar ug tangtangon kung ang channel magsira. Hangtud nga sirado ang channel, ang kliyente mahimong madiskonekta / makonektar, apan sulod lamang sa samang koneksyon. Kung ang parameter exclusive set, dayon ang parameter autoDelete walay epekto.

Features:

• kung adunay usa ka mubo nga pahulay sa komunikasyon, mawala ang mga mensahe nga wala pa maabot sa konsumidor
• mahimo nimong makuha ang panghitabo binding churn. Ang panghitabo mahitabo kung ang gidaghanon sa mga operasyon sa paghimo/pagtangtang sa mga pila ug mga pagbugkos moabot sa dako kaayo nga kantidad. Sa cluster mode, ang ingon nga dagan sa mga operasyon mokaylap sa tanan nga mga node ug maghimo usa ka dako nga karga. Kini nga proseso mahimong ma-optimize pinaagi sa pagkontrol sa gidaghanon sa mga subskripsyon

Permanenteng mga pila

Kung ang paghimo sa pila mahitabo sa set parameter durable, unya ang ingon nga pila nagluwas sa iyang estado ug gipahiuli human sa server/broker restart. Kini nga pila maglungtad hangtod ang mando gitawag Queue.Delete.

Daghang Anaa nga Pila

Ang mga pila sa HA nanginahanglan usa ka palibot nga kumpol sa RabbitMQ. Sa cluster mode, ang tanang impormasyon bahin sa exchangers, queues, bindings ug consumers makopya sa tanang node.

Kung ang usa ka mensahe gi-post sa usa ka HA queue, kini gitipigan sa matag node nga iya sa HA queue. Kung ang usa ka mensahe magamit sa usa sa mga node, ang tanan nga mga kopya sa kana nga mensahe mapapas sa ubang mga node.

Ang mga pila sa HA mahimong mosangkap sa tanan nga mga node sa usa ka cluster, o indibidwal nga mga node lamang.

Features:

• ang paggamit sa HA queues moresulta sa performance penalties. Kung magbutang usa ka mensahe sa pila ka pila sa HA o kung mogamit usa ka mensahe gikan sa pila nga HA, kinahanglan nga mag-coordinate ang RabbitMQ sa tanan nga mga node (kasagaran igo ang 2-3 nga mga node)

Paghimo og pila

Ang pila gihimo gamit ang synchronous RPC hangyo sa server. Ang hangyo gihimo gamit ang pamaagi Queue.Declare, gitawag nga adunay mga parameter:

• ngalan sa pila
• ubang mga kapilian

Usa ka pananglitan sa paghimo og pila gamit ang RabbitMQ.Kliyente:

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

• queue mao ang ngalan sa pila nga gusto natong buhaton. Ang ngalan kinahanglang talagsaon ug dili mahimong pareho sa ngalan sa sistema sa pila
• durable - kung tinuod, unya ang pila luwasa ang imong estado ug gipahiuli human sa server/broker restart
• exclusive - kung tinuod, ang pila magtugot lamang sa usa ka konsumidor nga magkonektar
• autoDelete - kung tinuod, nan ang pila nakakuha sa abilidad awtomatik nga tangtangon ang imong kaugalingon
• arguments mga opsyonal nga argumento. Sa ubos atong analisahon ang mas detalyado.

argumento

• x-message-ttl(x-message-time-to-live) - nagtugot kanimo sa pagtakda sa oras sa expiration sa mensahe sa milliseconds. Kung ang paghimo sa pila mahitabo sa kantidad sa argumento nga set x-message-ttl, unya ang ingon nga pila mahimong awtomatik nga dili iapil ang mga mensahe nga na-expire na. Pagtakda sa bili sa usa ka argumento x-message-ttl nagtakda sa pinakataas nga edad para sa tanang mensahe niini nga pila. Paghimo sa ingon nga usa ka pila makatabang sa pagpugong sa outdated nga impormasyon. Mahimo kining gamiton sa mga sistema sa real-time. Kung ang pila diin ang tigbaylo alang sa gisalikway nga mga mensahe gitakda, itakda ang kantidad sa argumento x-message-ttl, unya gisalikway ang mga mensahe niini nga pila magsugod nga adunay usa ka kinabuhi.
• x-expires — nagtakda sa kantidad sa milliseconds pagkahuman ang pila matangtang. Ang usa ka pila mahimo ra ma-expire kung wala kini mga subscriber. Kung ang mga subscriber konektado sa pila, kini mahimong awtomatiko nga mapapas lamang kung ang tanan nga mga subscriber motawag Basic.Cancel o idiskonekta. Ang usa ka pila mahimo ra ma-expire kung wala kini gihangyo Basic.Get. Kung dili, ang kasamtangan nga kantidad sa oras sa pagkinabuhi nga setting i-reset sa zero, ug ang pila dili na awtomatiko nga mapapas. Usab walay garantiya kung unsa ka paspas ang usa ka pila nga mapapas pagkahuman sa tibuok kinabuhi niini.
• x-max-length — nagtakda sa kinatas-ang gidaghanon sa mga mensahe sa pila. Kung ang gidaghanon sa mga mensahe sa pila magsugod nga molapas sa labing kadaghan nga numero, nan ang mga labing karaan magsugod sa pagtangtang.

• x-max-lenght-bytes - nagtakda sa pinakataas nga gitugot nga kinatibuk-ang payload nga gidak-on sa mga mensahe sa pila. Kung ang gitakda nga kantidad milapas (ang pila nag-awas sa sunod nga pagmantala sa mensahe), ang labing karaan nga mga mensahe mapapas.
• x-overflow - Kini nga argumento gigamit aron ipahiangay ang pamatasan ingon usa ka sangputanan sa pag-awas sa pila. Duha ka kantidad ang magamit: drop-head (default) ug reject-publish. Kung pilion drop-head, unya ang labing karaan nga mga mensahe mapapas. Kung pilion reject-publish, unya ang pagdawat sa mga mensahe masuspinde
• x-dead-letter-exchange - nagtino sa pagbinayloay diin ang mga gilabay nga mensahe nga wala gi-requeu gipadala
• x-dead-letter-routing-key - nagtakda ug opsyonal nga routing key para sa mga gisalikway nga mensahe
• x-max-priority - makapahimo sa paghan-ay pinaagi sa prayoridad sa pila nga adunay labing taas nga prayoridad nga kantidad nga 255 (RabbitMQ nga mga bersyon 3.5.0 ug mas taas). Gipiho sa numero ang labing taas nga prayoridad nga suportahan sa pila. Kung ang argumento wala gitakda, ang pila dili magpadayon sa prayoridad sa mensahe.
• x-queue-mode - nagtugot kanimo sa pagbalhin sa pila sa tapulan mode. Niini nga mode, daghang mga mensahe kutob sa mahimo ang tipigan sa disk. Ang paggamit sa RAM gamay ra. Kung dili kini mabutang, ang pila magtipig sa mga mensahe sa memorya aron mahatud ang mga mensahe sa labing madali.
• x-queue-master-locator - kung kita adunay usa ka cluster, nan mahimo natong itakda ang master queue
• x-ha-policy - gigamit sa paghimo sa HA nga mga pila ug gitino kung giunsa ang pag-apod-apod sa mensahe sa mga node. Kung gitakda sa all, unya ang mensahe itago sa tanan nga mga node. Kung gitakda sa nodes, unya ang mensahe itago sa pipila ka mga cluster node
• x-ha-nodes - nagtakda sa mga node diin mahisakop ang usa ka pila HA

Kung maghimo ug pila lagmit, unya ang server magpadala sa kliyente og usa ka synchronous RPC tubag Queue.DeclareOk. Kung maghimo ug pila dili mahimo (adunay pagdumili sa hangyo Queue.Declare) unya ang channel mosira server nga adunay mando Channel.Close ug ang kliyente makakuha og eksepsiyon OperationInterruptedException, nga adunay sulod nga error code ug ang paghulagway niini.

Hinumdumi Queue.Declare nga adunay parehas nga mga parameter ibalik ang mapuslanon nga kasayuran bahin sa kana nga pila. Pananglitan, ang kinatibuk-ang gidaghanon sa mga mensahe nga naghulat niini nga pila ug ang kinatibuk-ang gidaghanon sa mga konsumidor nga nag-subscribe niini.

Hagit Queue.Declare ubos sa mga kredensyal sa usa ka tiggamit nga wala matudlo sa gikinahanglan nga mga katungod isira ang channel gamit ang sugo Channel.Close ug ang kliyente makakuha og eksepsiyon OperationInterruptedException, nga maglangkob sa error code 403 ug deskripsyon niini.

Pagkahuman sa pila nga wala’y trabaho sa >= 10 segundos, kini moadto sa sleep mode, nga nagtawag sa GC sa pila, nga miresulta sa usa ka mahinungdanon nga pagkunhod sa panumduman nga gikinahanglan alang sa maong pila.

Paghimo og Queue pinaagi sa GUI

Adto sa admin panel RabbitMQ ubos sa user guest (username: guest ug password: guest). Palihug timan-i nga ang user guest mahimo ra magkonektar gikan sa localhost. Karon adto ta sa tab Queues ug i-klik ang Add a new queue. Pun-a ang mga kabtangan:

Sa diha nga kita mosulod sa tanan nga gikinahanglan nga data ug i-klik sa Add queues, ang pila makita sa kinatibuk-ang listahan.

Ang pag-klik sa ngalan sa pila magpakita sa detalyado nga kasayuran niini. Dinhi mahimo nimong i-configure ang pagbugkos tali sa pagbinayloay ug pila, tan-awa ang lista consumers, pagmantala/pagdawat og mga mensahe, pagtangtang sa pila ug tan-awa ang mga estadistika.

Paghimo sa usa ka Pagbugkos

Ang pagbugkos gihimo gamit ang usa ka synchronous RPC hangyo sa server. Ang hangyo gihimo gamit ang pamaagi Queue.Bind, gitawag nga adunay mga parameter:

• ngalan sa pila
• ngalan sa exchange point
• ubang mga kapilian

Usa ka pananglitan sa paghimo sa usa ka pagbugkos gamit RabbitMQ.Kliyente:

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

• queue - ngalan sa pila
• exchange - ngalan sa exchanger
• routingKey - routing key
• arguments - opsyonal nga mga argumento

Kung maghimo usa ka pagbugkos lagmit, unya ang server magpadala sa kliyente og usa ka synchronous RPC tubag Queue.BindOk.

Paghimo ug Pagbugkos Pinaagi sa GUI

Adto sa admin panel RabbitMQ ubos sa user guest (username: guest ug password: guest). Palihug timan-i nga ang user guest mahimo ra magkonektar gikan sa localhost. Karon adto ta sa tab Queues ug i-klik ang pila my_queue. Pun-a ang mga natad sa seksyon bindings:

Sa higayon nga kita mosulod sa tanan nga gikinahanglan nga data ug i-klik sa Bind, ang pagbugkos ipakita sa kinatibuk-ang listahan:

code

Niini nga seksyon, among ihulagway ang pila ug nagbugkos sa C # code, ingon nga kinahanglan namon nga maghimo usa ka librarya. Tingali kini mapuslanon alang sa panglantaw.

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; }
    }

Source: www.habr.com