Releases: kovalevdmv/1CRabbitMQ
0.9.91: Update README.md
вер. 0.9.91. Добавлена возможность указывать аргументы для очередей и потребителей. Через аргументы можно делать более гибкую настройку. В обработке РаботаСОчередямиТипаStream пример как через указание агрумента "x-queue-type" работать с очередями как с потоками (stream) аналогично как в apache kafka. Про возможные аргументы для очередей и потребителей (comsumers) можно почитать на офф сайте RMQ https://www.rabbitmq.com/docs/use-rabbitmq
Пример с стайта (объявление очереди и указание для нее аргументов):
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-queue-type", "stream");
// maximum stream size: 20 GB
arguments.put("x-max-length-bytes", 20_000_000_000);
// size of segment files: 100 MB
arguments.put("x-stream-max-segment-size-bytes", 100_000_000);
// size of stream bloom filter: 32
arguments.put("x-stream-filter-size-bytes", 32);
channel.queueDeclare(
"my-stream",
true, // durable
false, false, // not exclusive, not auto-delete
arguments
);
В данным случае arguments из примера будет указыватья аналогично для компоненты
АргументыОчереди = Новый Массив;
АргументыОчереди.Добавить(КлиентRMQ.НовыйАргумент("x-queue-type","stream"));
Очередь = КлиентRMQ.ОбъявитьОчередь(Канал, "node1", "nomra", КлиентRMQ.Exchange_direct(), Истина,,,АргументыОчереди);
Аналогично любые другие аргументы. Аргументы это пара ключ/значение. Подробнее в примере Обработка.РаботаСОчередямиТипаStream
Так же добавлена возможность указывать для канала количество предзагруженных не подтвержденных сообщений.
Это полезно если в очереди очень много сообщений. а клиент обрабатывает их медленно. Тогда для экономии памяти на клиенте можно указать н.п.
КоличествоПредзагруженныхНеПодтвержденныхСообщений = 10;
Канал = КлиентRMQ.СоздатьКанал(Подключение,,КоличествоПредзагруженныхНеПодтвержденныхСообщений);
Тогда RMQ сервер будет слать только 10 не подтвержденных сообщений. И после подтверждения будет присылать новые. Если сообщения не подтверждать, то сервер не будет присылать сообщения. Поэтому если нужно обработать все сообщения без подтверждения, этот параметр не нужно использовать.
Так же он необходим для работы с очередями в режиме потока (Strem) без установки этого параметра RMQ сервер не разрешит установить соединение с очередью типа Stream, так как поток не очищаются и хранит все сообщения (модель работы как в kafka) и если не указать размер сообщений, серверу пришлось бы прислать все сообщения которые накопились в потоке.
0.9.90: Update README.md
Добавлен метод ОтклонитьСообщение() и возможность чтения заголовка сообщения x-death.
Сценарий где это можно применить:
Например необходимо иметь возможность отклонить сообщение, но не просто НЕ подтвердить, а сообщение должно автоматически переместиться в мертвую очередь, где будут находится сообщения, которые не возможно обработать. При этом, при чтении этого сообщения из мертвой очереди, сервер RMQ добавит к сообщению заголовок x-death где будет информация о том из какой очереди оно переместилось в мертвую (после отклонения) и сколько раз (и прочая информация). Чтобы сообщение автоматически перемещалось в мертвую очередь, в настройках исходной очереди нужно в аргументе x-dead-letter-exchange указать точку обмена. Подробнее этот сценарий описан в справке в обработке ПримерОтклоненияСообщений в демо конфигурации. При этом сообщение при отклонении может так же вернуть и в исходную очередь. За это поведение отвечает параметр ВернутьВИсходнуюОчередь в методе ОтклонитьСообщение(). Если ВернутьВИсходнуюОчередь=Истина, то сообщение вернется в исходную очередь. Если ВернутьВИсходнуюОчередь=Ложь, то: если для исходной очереди не указан аргумент x-dead-letter-exchange, то удалится, если указан, то переместится в мертвую очередь.
0.9.89: Update README.md
Добавлена справка по решению возможных проблем при подключении компоненты.
Для просмотра справки нужно открыть обработку КлиентRMQ и нажать F1
0.9.88
Добавлена возможность указывать пользовательские заголовки для сообщений.
Пример в обработке ИспользованиеЗаголовковСообщений
При отправке
Заголовки = КлиентRMQ.ДобавитьЗаголовок("Заголовок1", "ЗначениеЗаголовка1");
Заголовки = КлиентRMQ.ДобавитьЗаголовок("Заголовок2", "ЗначениеЗаголовка2", Заголовки);
Заголовки = КлиентRMQ.ДобавитьЗаголовок("Заголовок3", "ЗначениеЗаголовка3", Заголовки);
Ответ = КлиентRMQ.ОпубликоватьСообщение(Канал, Данные, "nomra", КлиентRMQ.Exchange_direct(),,,,Заголовки);
При получении
Для каждого текЗаголовок Из ДанныеСообщения.Заголовки Цикл
Сообщить(СтрШаблон("%1:%2", текЗаголовок.Ключ, текЗаголовок.Значение));
КонецЦикла;
0.9.87: Update README.md
Добавлена возможность указывать таймаут при ожидании новых сообщений в очереди. Таймаут указывается в параметрах
получателя.
// Создать получателя, через которого будут получаться сообщения из очереди
Таймаут = 5; // таймаут при ожидании сообщения (в сек). Если = 0, таймаута нет.
Получатель = КлиентRMQ.СоздатьПолучателя(Канал, "node1", "ПолучательНоменклатуры",,,,,Таймаут);
0.9.86: Update README.md
Изменен тип подключения компоненты для linux. Теперь для linux компонента так же подключается в изолированном режиме. В случае изолированного режима компонента работает в отдельном процессе addnhost64 (не в rphost как в случае не изолированного) и в случае ошибки не приведет к падению rphost.
0.9.85
Добавлена возмжоность указывать флаги для потребителя:
no_ack Булево
Истина - автоматическое подтверждение получения сообщений
Ложь - требуется явное подтверждение получения каждого сообщения
exclusive Булево
Истина - только один получатель может быть подключен к очереди
Ложь - разрешено подключение нескольких получателей
no_local Булево
Истина - не получать сообщения, опубликованные через это же соединение
Ложь - получать все сообщения независимо от источника
nowait Булево
Истина - не ждать подтверждения от сервера при создании получателя
Свойства указываются в конструкторе потребителя
// Функция - Создать получателя для RabbitMQ
//
// Параметры:
// Канал - Число - Идентификатор открытого канала RabbitMQ
// ИмяОчереди - Строка - Имя очереди, из которой будут читаться сообщения
// ТэгПолучателя - Строка - Любой строковый тег получателя
// no_ack - Булево - Истина - автоматическое подтверждение получения сообщений
// Ложь - требуется явное подтверждение получения каждого сообщения
// exclusive - Булево - Истина - только один получатель может быть подключен к очереди
// Ложь - разрешено подключение нескольких получателей
// no_local - Булево - Истина - не получать сообщения, опубликованные через это же соединение
// Ложь - получать все сообщения независимо от источника
// nowait - Булево - Истина - не ждать подтверждения от сервера при создании получателя
// Ложь - ждать подтверждение создания получателя от сервера
//
// Возвращаемое значение:
// - Структура с данными созданного получателя
//
Функция СоздатьПолучателя(Канал, ИмяОчереди, ТэгПолучателя = "", no_ack = Ложь, exclusive = Ложь, no_local = Ложь, nowait = Ложь) Экспорт
Если ЭкземплярКомпоненты = Неопределено Тогда
Результат = ПодключитьКомпоненту();
Если ЭтоОшибка(Результат) Тогда
Возврат Результат;
КонецЕсли;
КонецЕсли;
ОпцииПолучателя = ОпцииПолучателя(Канал, ИмяОчереди, ТэгПолучателя, no_ack, exclusive, no_local, nowait);
Получатель = ЭкземплярКомпоненты.ДобавитьПолучателяСообщений(Сериализовать(ОпцииПолучателя));
Возврат Десериализовать(Получатель);
КонецФункции
0.9.8
Добавлена возможность использовать стандартные свойства сообщения (которые определяются самим протоколом AMQP):
content_type, content_encoding, priority, correlation_id, reply_to, expiration, message_id, timestamp, type, user_id, app_id, cluster_id. Пример использования этих свойств в обработке ИспользованиеСвойствСообщений.
Один из сценариев использования этих свойств — это реализация RPC через брокеры сообщений.
Общая концепция реализации RPC через RabbitMQ:
Предположим, что есть клиент и сервер; необходимо вызвать удаленные процедуры с клиента на сервере и вернуть результат (RPC).
- Создается очередь, например,
rpc_queue, в которую будут поступать запросы от клиента и которую будет прослушивать сервер. - Клиент отправляет запрос в эту очередь, например, для выполнения какого-либо расчета.
- В сообщение клиент добавляет свойство
reply_toс названием очереди, в которую сервер должен отправить ответ. - После отправки запроса клиент прослушивает очередь, указанную в свойстве
reply_to. - Сервер, получив сообщение с запросом из очереди rpc_queue, обрабатывает его и отправляет ответ в очередь, указанную в свойстве
reply_to(указанную клиентом). - Клиент получает ответ и обрабатывает его соответствующим образом.
Если необходимо сязать запрос и ответ, то клиент может так же указать УИД запроса в свойстве correlation_id, сервер в ответном сообщение скопирует это свойство, таким образом клиет будет знать на какой именно запрос ему ответил сервер.
Заполнение и работа со свойствами на стороне клиента и сервера реализуется вручную, RMQ не вмешивается в этот процесс, он на уровне протокола AMQP предоставляет готовые свойства. Но есть свойства которые обрабатывает RMQ, н.п. если указать время жизни в свойстве expiration, то по истечении этого времени RMQ удалить сообщение из очереди.
Общую концепцию так же можно почать в туториале на сайте rqm
0.9.7
Исправлена ошибка подключения компоненты в linux
0.9.6
Добавлена возможность прерывать ожидание нового сообщения из очереди. Так как ожидание в функции КлиентRMQ.СледующееСообщение(Получатель) это блокирующее ожидание, т.к. оно блокирует поток выполнения 1С до момента когда в очереди появится сообщение, а иногда необходимо прервать ожидание, то можно использовать пару методов: 1. КлиентRMQ.ПолучитьСписокПолучателейСообщений() который вернет массив всех получателей и 2. КлиентRMQ.ОстановитьОжиданиеСообщения(Получение). Так это можно делать интерактивно, через интерфейс обработки КлиентRMQ, там же можно посмотреть как это работает, для программного прерывания ожидания сообщения.
Подробнее в видео обзоре этой функции