RabbitMQ. Част 3. Разбиране на Queues и Bindings

Опашка (опашка) — структура от данни на диска или в оперативната памет, която съхранява връзки към съобщения и предоставя техни копия consumers (консуматорите). Опашка представлява Erlang процес състояние (където могат да се кешират и самите съобщения). 1 хиляда опашки могат да заемат около 80Mb.

Връзка (привързване) — правило, което информира обменника в коя от опашките трябва да попадат съобщенията.

Съдържание

• RabbitMQ. Част 1. Въведение. Erlang, AMQP и RPC
• RabbitMQ. Част 2. Разглеждаме Exchanges
• RabbitMQ. Част 3. Разбиране на Queues и Bindings
• RabbitMQ. Част 4. Разглеждаме какво представляват съобщенията и фреймовете
• RabbitMQ. Част 5. Производителност на публикуване и потребление на съобщения
• RabbitMQ. Част 6. Преглед на модулите Federation и Shovel
• RabbitMQ. Част 7. Подробно за Connection и Chanel
• RabbitMQ. Част 8. RabbitMQ в .NET
• RabbitMQ. Част 9. Мониторинг

Временни опашки

Ако създаването на опашка се извършва с установен параметър autoDelete, то такава опашка придобива способността автоматично да се изтрива. Такива опашки обикновено се създават в момента на свързване на първия клиент и се изтриват, когато всички клиенти се отсъединят.

Ако създаването на опашка се извършва с установен параметър ексклузивна, то такава опашка разрешава свързването само на един консуматор и се изтрива, ако каналът се затвори. Докато каналът не се затвори, клиентът може да се отсъединява/свързва, но само в рамките на същата връзка. Ако параметърът ексклузивна е установен, то параметърът autoDelete няма никакъв ефект.

Характеристики:

• при краткотрайно прекъсване на връзката ще губим съобщения, които все още не са стигнали до консуматора.
• можем да наблюдаваме явлението binding churn. Явлението възниква, когато броят на операциите за създаване/изтриване на опашки и връзки достигне много големи стойности. В клъстерен режим този поток от операции ще се разпространява по всички възли и ще създаде голямо натоварване. Този процес може да бъде оптимизиран чрез контрол на броя на абонаментите.

Постоянни опашки

Ако създаването на опашка се извършва с установен параметър durable, то такава опашка запазва състоянието си и възстановява се след рестартиране на сървъра/брокера. Тази опашка ще съществува до момента, в който не бъде извикана команда Queue.Delete.

Сигурни опашки

Опашките 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, опашката ще запазва състоянието си и се възстановява след рестартиране на сървъра/брокера
• ексклузивна — ако е 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-length-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 опашки и определя как съобщението ще се разпространява между възлите. Ако е зададена стойност всички, тогава съобщението ще се запазва на всички възли. Ако е зададена стойност nodes, тогава съобщението ще се запазва на определени възли от клъстера
• x-ha-nodes — задава възлите, към които ще се отнася определена опашка HA

Ако създаването на опашка е възможно, то сървърът ще изпрати синхронен на клиента RPC отговор Queue.DeclareOk. Ако създаването на опашка не е възможно (възникнало е отказване на заявката Queue.Declare), то каналът ще бъде затворен от сървъра с команда Channel.Close и клиентът ще получи изключение OperationInterruptedException, което ще съдържа код на грешка и нейното описание.

Повторно извикване Queue.Declare с аналогични параметри ще върне полезна информация за тази опашка. Например, общият брой на съобщенията в изчакване в дадената опашка и общият брой на потребителите, подписани за нея.

Извикване Queue.Declare с данните за вход на потребителя, на когото не са назначени необходимите права ще затвори канала с командата Channel.Close и клиентът ще получи изключение OperationInterruptedException, което ще съдържа код на грешка 403 и нейното описание.

След като опашката е в покой за >= 10 секунди, тя попада в спящ режим, което предизвиква GC в опашката, водещо до значително намаляване на паметта, необходима за тази опашка.

Създаване на опашка през графичен интерфейс

Влизаме в административната панела RabbitMQ с потребителя guest (потребителско име: guest и парола: guest). Обърнете внимание, че потребителят guest може да се свързва само от локалния хост. Сега преминаваме към таба Queues и натискаме на Add a new queue. Попълваме свойствата:

След като въведем всички необходими данни и натиснем на Add queues, опашката ще се появи в общия списък.

Щракването върху името на опашката ще покаже детайлна информация за нея. Тук можете да настроите свързването между обмена и опашката, да видите списъка consumers, да публикувате/получите съобщения, да изтриете опашката и да видите статистика.

Създаване на свързване

Създаването на свързване става чрез синхронно RPC запитване към сървъра. Запитването се извършва чрез метода Queue.Bind, извикван с параметри:

• името на опашката
• име на точка на обмен
• други параметри

Пример за създаване на свързване чрез RabbitMQ.Client:

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

• queue — име на опашка
• exchange — име на обменник
• routingKey — ключ за маршрутизация
• arguments — аргументи по избор

Ако създаването на свързване е възможно, то сървърът ще изпрати синхронен на клиента RPC отговор Queue.BindOk.

Създаване на свързване през графичен интерфейс

Влизаме в административната панела RabbitMQ с потребителя guest (потребителско име: guest и парола: guest). Обърнете внимание, че потребителят guest може да се свързва само от локалния хост. Сега преминаваме към таба Queues и щракваме на опашката my_queue. Попълваме полетата в раздела bindings:

След като въведем всички необходими данни и натиснем на Bind, свързването ще се появи в общия списък:

Code

В този раздел ще опишем опашката и свързването с код на C#, така както ако се нуждаехме от разработване на библиотека. Може би това ще бъде полезно за възприемането.

публичен интерфейс IQueue
    {        
        низ Name { получаване; }

         <резюме>
             Ако се установи на true, то queue ще бъде постоянно. 
             Ще се съхранява на диска и ще може 
             да преживее рестартиране на сървъра/брокера. 
             Ако стойността е false, то queue е временно и ще бъде изтрито, 
             когато сървърът/брокерът бъде рестартиран
         </резюме>
        бул IsDurable { получаване; }

         <резюме>
             Ако стойността е true, то 
             такава опашка ще разреши свързването 
             само на един потребител
         </резюме>
        бул IsExclusive { получаване; }

         <резюме>
             Автоматично изтриване. 
             Опашката ще бъде изтрита, когато всички клиенти се отвърнат.
         </резюме>
        бул IsAutoDelete { получаване; }

         <резюме>
             Незадължителни аргументи
         </резюме>
        IDictionary<низ, обект> Аргументи { получаване; }
    }

публичен клас Queue : IQueue
    {
        публичен Queue(
             низ име, 
             бул isDurable = true, 
             бул isExclusive = false, 
             бул isAutoDelete = false, 
             IDictionary<низ, обект> аргументи = null)
        {
            Name = име ??
                хвърля ново ArgumentNullException(име, $"{име} не трябва да бъде null");

            IsDurable = isDurable;
            IsExclusive = isExclusive;
            IsAutoDelete = isAutoDelete;
            Arguments = аргументи ?? нова Dictionary<низ, обект>();
        }

        публичен низ Name { получаване; }
        публичен бул IsDurable { получаване; }
        публичен бул IsExclusive { получаване; }
        публичен бул IsAutoDelete { получаване; }
        публичен IDictionary<низ, обект> Arguments { получаване; }
    }

публичен статичен клас QueueMode
    {       
        публична константа низ Default = "default";
         <резюме>
             Ленив режим. Ленивият режим ще накара да се съхранява 
             колкото е възможно повече съобщения на диска, за да се намали 
             използването на оперативната памет
         </резюме>
        публична константа низ Lazy = "lazy";
    }

публичен интерфейс IBinding
    {
         <резюме>
             Обменник, който ще бъде свързан към привързаност
         </резюме>
        IExchange Exchange { получаване; }

         <резюме>
             Ключ за маршрутизиране
         </резюме>
        низ RoutingKey { получаване; }

         <резюме>
             Незадължителни аргументи
         </резюме>
        IDictionary<низ, обект> Аргументи { получаване; }
    }

публичен клас Binding : IBinding
    {
        публичен Binding(
             IExchange exchange, 
             низ routingKey, 
             IDictionary<низ, обект> аргументи)
        {
            Exchange = обмен;
            RoutingKey = routingKey;
            Arguments = аргументи;
        }

        публичен IExchange Exchange { получаване; }
        публичен низ RoutingKey { получаване; }
        публичен IDictionary<низ, обект> Arguments { получаване; }
    }

Източник: habr.com