RabbitMQ. Часть 3. Разбираемся с очередями и привязками

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

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

Съдържание

• RabbitMQ. Часть 1. Введение. Erlang, AMQP и RPC
• RabbitMQ. Част 2. Разглеждаме Exchanges
• RabbitMQ. Часть 3. Разбираемся с очередями и привязками
• RabbitMQ. Часть 4. Разбираемся с тем, что такое сообщения и фреймы
• RabbitMQ. Часть 5. Производительность публикации и потребления сообщений
• RabbitMQ. Часть 6. Обзор модулей Federation и Shovel
• RabbitMQ. Часть 7. Подробно про Connection и Chanел
• 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-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 в опашката, което води до значително намаляване на паметта, необходима за тази опашка.

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

Влизаме в администрацията RabbitMQ като потребител гость (username: гость и password: гость). Обърнете внимание, че потребителят гость може да се свързва само от локалния хост. Сега преминаваме на таба Опашки и кликваме на Добавяне на нова опашка. Попълваме свойствата:

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

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

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

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

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

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

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

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

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

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

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

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

Code

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

public interface IQueue
    {        
        string Name { get; }

        // 

        //     Ако се зададе true, опашката ще бъде постоянна. 
        //     Тя ще се съхранява на диска и ще може 
        //     да оцелее при рестартиране на сървъра/брокера. 
        //     Ако е false, опашката е временна и ще бъде изтрита, 
        //     когато сървърът/брокерът бъде рестартиран
        // 
        bool IsDurable { get; }

        // 

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

        // 

        //     Автоматично изтриване. 
        //     Опашката ще бъде изтритa, когато всички клиенти се отсъединят.
        // 
        bool IsAutoDelete { get; }

        // 

        //     Опционални аргументи
        // 
        IDictionary Arguments { get; }
    }

public class Queue : IQueue
    {
        public Queue(
             string name, 
             bool isDurable = true, 
             bool isExclusive = false, 
             bool isAutoDelete = false, 
             IDictionary arguments = null)
        {
            Name = name ??
                throw new ArgumentNullException(name, $"{name} не трябва да бъде null");

            IsDurable = isDurable;
            IsExclusive = isExclusive;
            IsAutoDelete = isAutoDelete;
            Arguments = arguments ?? new Dictionary();
        }

        public string Name { get; }
        public bool IsDurable { get; }
        public bool IsExclusive { get; }
        public bool IsAutoDelete { get; }
        public IDictionary Arguments { get; }
    }

public static class QueueMode
    {       
        public const string Default = "default";
        // 

        //     Ленив режим. Ленивият режим ще принуди системата да запазва 
        //     колкото се може повече съобщения на диска, за да намали 
        //     използването на оперативната памет
        // 
        public const string Lazy = "lazy";
    }

public interface IBinding
    {
        // 

        //     Обменник, който ще бъде свързан с привързването
        // 
        IExchange Exchange { get; }

        // 

        //     Ключ за маршрутизиране
        // 
        string RoutingKey { get; }

        // 

        //     Необходими аргументи
        // 
        IDictionary Arguments { get; }
    }

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

        public IExchange Exchange { get; }
        public string RoutingKey { get; }
        public IDictionary Arguments { get; }
    }

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