AMQP
Общее описание
Блок AMQP позволяет осуществлять обмен сообщениями по протоколу AMQP с использованием клиента Apache Qpid.
AMQP (Advanced Message Queuing Protocol) — открытый стандарт прикладного уровня для передачи сообщений между компонентами распределенной системы. Протокол обеспечивает надежную, гарантированную доставку сообщений, поддерживает маршрутизацию по правилам и гарантирует транзакционность операций.
Блок AMQP может использоваться в качестве:
-
блока отправителя
-
блока получателя
Настройка блока
В данном разделе описаны параметры блока AMQP, которые необходимо заполнить при его настройке.
Базовые параметры
| Название в UI | Название атрибута | Описание | Значение по умолчанию | Тип данных |
|---|---|---|---|---|
| Dynamic Endpoint | dynamicEndpoint | Параметр позволяет включить встроенную функциональность динамического определения вызываемой конечной точки. При включении параметра структура блока меняется таким образом, что в коде вместо определения to проставляется toD, а значение самого вызываемого ресурса может собираться из переменной, например ${header.foo}. | True | Логическое значение |
| Connection | connection | Обязательный параметр Этот параметр позволяет выбрать предсозданное подключение или создать новое с помощью визарда Connection Manager. | Список | |
| Description | description | Краткое описание блока. | Строка | |
| Destination Name | destinationName | Обязательный параметр Задает "адрес" доставки AMQP‑сообщений. Он определяет конкретную очередь (queue) или тему (topic) на брокере, куда отправляются или откуда забираются сообщения. Правильная настройка этого параметра критически важна для корректной работы интеграционных потоков. | Строка | |
| Client Id | clientId | Идентификатор клиента – это указанная пользователем строка, отправляемая в каждом запросе для отслеживания вызовов. Она должна логически идентифицировать приложение, отправляющее запрос. Введенное значение должно быть уникальным в системе. | Строка | |
| Disable Reply To | disableReplyTo | Управляет созданием заголовка ReplyTo в исходящем сообщении. Если установлено значение True, компонент отключает автоматическое формирование заголовка Reply‑To, который обычно указывает адрес для ответа (например, очередь или endpoint). Это предотвращает неявную обратную связь и позволяет явно контролировать маршрутизацию ответов. Если False, заголовок Reply‑To формируется автоматически на основе контекста потока (например, копируется из входящего сообщения или задается по умолчанию). | False | Логическое значение |
| Durable Subscription Name | durableSubscriptionName | Имя постоянной подписки (сохраняется на брокере при отключении клиента). Чтобы пользоваться этой опцией, необходимо задать значение параметру Client ID | Строка | |
| Jms Message Type | jmsMessageType | Позволяет принудительно использовать определенную реализацию jakarta.jms.Message для отправки сообщений JMS. По умолчанию тип сообщения, который будет использоваться, определяется исходя из типа In body. Возможные значения:
| Список | |
| Reply To | replyTo | Задает явного адресата ReplyTo, переопределяя значение Message.getJMSReplyTo(), полученное в исходном сообщении. | Строка | |
| Test Connection On Startup | testConnectionOnStartup | Если True, компонент проверяет доступность AMQP‑сервера при запуске потока. Гарантирует, что соединение установлено до начала обработки сообщений. При ошибке — поток не стартует. | False | Логическое значение |
| Delivery Delay | deliveryDelay | Устанавливает задержку доставки, которая будет использоваться для отправки вызовов JMS. Для этого параметра требуется брокер, совместимый с JMS 2.0. | -1 | Целое число |
| Delivery Mode | deliveryMode | Указывает используемый режим доставки. Доступные значения определены в jakarta.jms.Delivermode. Возможные значения:
| Список | |
| Delivery Persistent | deliveryPersistent | Задает, используется ли непрерывная доставка по умолчанию. | True | Логическое значение |
| Explicit Qos Enabled | explicitQosEnabled | Определяет, следует ли использовать при отправке сообщений свойства Delivery Mode, Priority или Time To Live. Параметры Delivery Mode, Priority и Time To Live применяются к текущей конечной точке. | False | Логическое значение |
| Format Date Headers To Iso8601 | formatDateHeadersToIso8601 | Определяет, следует ли форматировать даты JMS в соответствии со стандартом ISO 8601. | False | Логическое значение |
| Preserve Message Qos | preserveMessageQos | Включает отправку сообщений с использованием настроек QoS, указанных в сообщении, а не настроек QoS на конечной точке JMS. Следующие три заголовка обрабатываются: JMSPriority, JMSDeliveryMode и JMSExpiration. Вы можете указать все или только некоторые из них. Если они не указаны, то будут применены значения из конечной точки. Таким образом, при использовании этой опции заголовки переопределяют значения из конечной точки. Опция Explicit Qos Enabled, напротив, будет использовать только параметры, установленные в конечной точке, а не значения из заголовка сообщения. | False | Логическое значение |
| Priority | priority | Приоритет сообщения при отправке (где 1 — самый низкий приоритет, а 9 — самый высокий). Для того чтобы этот параметр работал, также должна быть включена опция Explicit Qos Enabled. Возможные значения:
| 4 | Список |
| Reply To Concurrent Consumers | replyToConcurrentConsumers | Максимальное количество одновременных потребителей при использовании запроса/ответа через JMS. Также для управления динамическим увеличением/уменьшением масштабирования потоков применяется параметр Max Messages Per Task. | 1 | Целое число |
| Reply To Max Concurrent Consumers | replyToMaxConcurrentConsumers | Максимальное количество одновременных пользователей при использовании запроса/ответа через JMS. Также для управления динамическим увеличением/уменьшением масштабирования потоков применяется параметр Max Messages Per Task. | Целое число | |
| Reply To On Timeout Max Concurrent Consumers | replyToOnTimeoutMaxConcurrentConsumers | Указывает максимальное количество одновременных пользователей для продолжения маршрутизации по истечении времени ожидания при использовании запроса/ответа через JMS. | 1 | Целое число |
| Reply To Override | replyToOverride | Явное указание адресата ReplyTo в сообщении JMS, которое переопределяет параметр replyTo. Это полезно, если вы хотите переслать сообщение в удаленную очередь и получить ответное сообщение от адресата ReplyTo. | Строка | |
| Reply To Type | replyToType | Явное определение того, какой тип стратегии будет использоваться для очередей replyTo при выполнении запроса/ответа через JMS. По умолчанию будут использоваться временные очереди. Однако, если replyTo задан, то по умолчанию используется общий доступ. Этот параметр позволяет использовать эксклюзивные очереди вместо общих. Возможные значения:
| Список | |
| Request Timeout | requestTimeout | Время ожидания ответа при использовании InOut Exchange Pattern (в миллисекундах). Значение по умолчанию равно 20 секундам. Вы можете включить заголовок CamelJmsRequestTimeout, чтобы переопределить это значение тайм‑аута, настроенное для конечной точки, и, таким образом, задать индивидуальные значения тайм‑аута для каждого сообщения. Смотрите также параметр Request Timeout Checker Interval. | 20000 | Длинное целое число |
| Time To Live | timeToLive | При отправке сообщений указывает время жизни сообщения в миллисекундах. | -1 | Целое число |
| Error Handler Logging Level | errorHandlerLoggingLevel | Уровень логирования ErrorHandler по умолчанию для логирования неперехваченных исключений. Возможные значения:
| WARN | Список |
| Error Handler Log Stack Trace | errorHandlerLogStackTrace | Определяет, следует ли логировать трассировку стека или нет, с помощью обработчика ошибок по умолчанию. | True | Логическое значение |
| Transacted | transacted | Включает транзакционный режим. | False | Логическое значение |
| Transacted In Out | transactedInOut | Указывает, будут ли операции InOut (ответ на запрос) по умолчанию использоваться в режиме транзакций. | False | Логическое значение |
Расширенные параметры
| Название в UI | Название атрибута | Описание | Значение по умолчанию | Тип данных |
|---|---|---|---|---|
| Allow Additional Headers | allowAdditionalHeaders | Разрешает использование дополнительных заголовков, которые могут иметь недопустимые значения в соответствии со спецификацией JMS. Можно указать несколько имен заголовков через запятую — они будут использоваться как суффикс для сопоставления с подстановочными знаками. | Строка | |
| Allow Null Body | allowNullBody | Определяет, разрешено ли отправлять сообщения без текста. Если параметр имеет значение False, а текст сообщения равен null, генерируется исключение JMSException. | True | Логическое значение |
| Always Copy Message | alwaysCopyMessage | Если значение True, копия сообщения JMS создается всегда, когда оно передается производителю для отправки. Копирование необходимо в некоторых ситуациях — например, когда задано значение ReplyToDestinationSelectorName. Примечание: Если задан параметр ReplyToDestinationSelectorName, то значение AlwaysCopyMessage будет определяться как True. | False | Логическое значение |
| Correlation Property | correlationProperty | Используется при шаблоне обмена InOut: задает свойство JMS вместо стандартного JMSCorrelationID для корреляции сообщений. Если установлено, сообщения коррелируются исключительно по значению этого свойства, а JMSCorrelationID игнорируется. | Строка | |
| Disable Time To Live | disableTimeToLive | При включении отключает время жизни сообщения принудительно. | False | Логическое значение |
| Force Send Original Message | forceSendOriginalMessage | Если MapJmsMessage=False, при изменении заголовков (get/set) во время потока создается новое JMS‑сообщение для отправки новому получателю. Если параметр установлен в True, отправляется исходное полученное JMS‑сообщение. | False | Логическое значение |
| Include Sent JMSMessage ID | includeSentJMSMessageID | Применяется только при отправке получателю JMS с использованием InOnly («отправить и забыть»). При включении добавляет в данные потока фактический идентификатор JMSMessageID, использованный клиентом JMS при отправке сообщения. | False | Логическое значение |
| Lazy Start Producer | lazyStartProducer | Определяет, должен ли отправитель стартовать в отложенном режиме (при получении первого сообщения). Отложенный запуск полезен, если запуск отправителя может завершиться неудачей и привести к сбою при старте потока. При отложенном запуске сбой можно обработать во время маршрутизации сообщений с помощью обработчиков ошибок. Примечание: При обработке первого сообщения создание и старт отправителя могут занять время и увеличить общее время обработки. | False | Логическое значение |
| Reply To Cache Level Name | replyToCacheLevelName | Устанавливает уровень кэширования по имени для получателя ответа при выполнении запроса/ответа через JMS. Применяется только при использовании фиксированных очередей ответов (не временных).По умолчанию: CACHE_CONSUMER для эксклюзивного/общего доступа с заданным ReplyToSelectorName и CACHE_SESSION для общего доступа без ReplyToSelectorName.При использовании временных очередей CACHE_NONE не разрешен — следует использовать CACHE_CONSUMER или CACHE_SESSION. Возможные значения:
| Список | |
| Reply To Destination Selector Name | replyToDestinationSelectorName | Устанавливает селектор JMS с фиксированным именем. Используется для фильтрации собственных ответов от других при работе с общей очередью (если не используется временная очередь ответов). | Строка | |
| Stream Message Type Enabled | streamMessageTypeEnabled | Определяет, включен ли тип StreamMessage. Полезная нагрузка потокового типа (файлы, InputStream и т. д.) отправляется либо как BytesMessage, либо как StreamMessage. Параметр задает, какой тип использовать.По умолчанию используется BytesMessage (вся полезная нагрузка считывается в память). При включении полезная нагрузка считывается и записывается в StreamMessage частями. | False | Логическое значение |
| Allow Serialized Headers | allowSerializedHeaders | Определяет, включать ли сериализованные заголовки. Применяется, только если transferExchange=True. Требует, чтобы объекты были сериализуемыми. Несериализуемые объекты исключаются и регистрируются на уровне WARN. | False | Логическое значение |
| Artemis Streaming Enabled | artemisStreamingEnabled | Включает оптимизацию для режима потоковой передачи Apache Artemis. Снижает нагрузку на память при использовании Artemis с типами потоковых сообщений JMS. Должен быть включен только при использовании Apache Artemis. | False | Логическое значение |
| Async Start Listener | asyncStartListener | Определяет, запускать ли асинхронный листенер сообщений JmsConsumer при старте потока. Если JmsConsumer не может установить соединение с удаленным JMS‑брокером, он может заблокироваться при повторных попытках. Установка параметра в True разрешает запуск потоков, пока JmsConsumer подключается к брокеру в асинхронном режиме.Если соединение не удалось установить, регистрируется исключение на уровне WARN, получатель не получает сообщения — поток можно перезапустить для повторной попытки. | False | Логическое значение |
| Async Stop Listener | asyncStopListener | Определяет, следует ли останавливать листенер сообщений JmsConsumer асинхронно при остановке потока. | False | Логическое значение |
| Destination Resolver | destinationResolver | Подключаемый интерфейс стратегии для разрешения назначений JMS. | Строка | |
| Error Handler | errorHandler | Обработчик ошибок, вызываемый при возникновении неперехваченных исключений при обработке сообщения. По умолчанию исключения регистрируются на уровне WARN, если обработчик не задан. Можно настроить уровень журнала и регистрацию трассировок стека. | Строка | |
| Exception Listener | exceptionListener | Задает листенер исключений JMS, получающий уведомления о базовых исключениях JMS. | Строка | |
| Header Filter Strategy | headerFilterStrategy | Имя пользовательской стратегии HeaderFilterStrategy для фильтрации заголовков в сообщении Camel и обратно. | Строка | |
| Idle Consumer Limit | idleConsumerLimit | Ограничивает количество потребителей, которые могут простаивать одновременно. | 1 | Целое число |
| Idle Task Execution Limit | idleTaskExecutionLimit | Определяет предел числа запусков в режиме ожидания задачи получения без результата. При достижении предела задача завершается и передает получение другим выполняющимся задачам (в случае динамического планирования). | 1 | Целое число |
| Include All JMSXProperties | includeAllJMSXProperties | Определяет, следует ли включать все свойства с префиксом JMSX при преобразовании из JMS в контекст. При True включаются такие свойства, как JMSXAppID, JMSXUserID и т. д. Примечание: Неприменимо при использовании пользовательской стратегии HeaderFilterStrategy. | False | Логическое значение |
| Jms Key Format Strategy | jmsKeyFormatStrategy | Подключаемая стратегия кодирования и декодирования ключей JMS для соответствия спецификации. Предоставляются две реализации: default (безопасное использование точек и дефисов) и passthrough (ключ остается как есть). Можно предоставить собственную реализацию (обозначение #). Возможные значения:
| Список | |
| Map Jms Message | mapJmsMessage | Определяет, будет ли полученное JMS‑сообщение автоматически сопоставляться с подходящим типом полезной нагрузки (например, jakarta.jms.TextMessage в строку). | True | Логическое значение |
| Max Messages Per Task | maxMessagesPerTask | Количество сообщений для каждой задачи. Значение "-1" означает отсутствие ограничений. При использовании диапазона одновременных потребителей (например, min/max) параметр контролирует, насколько быстро количество потребителей будет уменьшаться при снижении нагрузки. | -1 | Целое число |
| Message Converter | messageConverter | Пользовательский преобразователь сообщений Spring (org.springframework.jms.support.converter.MessageConverter) для контроля сопоставления отправляемых/получаемых сообщений с jakarta.jms.Message. | Строка | |
| Message Created Strategy | messageCreatedStrategy | Стратегия, вызываемая при создании новых экземпляров объектов jakarta.jms.Message во время отправки сообщения JMS. | Строка | |
| Message Id Enabled | messageIdEnabled | При отправке указывает, следует ли добавлять идентификаторы сообщений (подсказка для JMS‑брокера). Если брокер принимает подсказку, идентификатор сообщения устанавливается в null; если игнорирует — присваивается уникальное значение. | True | Логическое значение |
| Message Listener Container Factory | messageListenerContainerFactory | Регистрационный идентификатор messageListenerContainerFactory для определения используемого org.springframework.jms.listener.AbstractMessageListenerContainer. | Строка | |
| Message Timestamp Enabled | messageTimestampEnabled | Указывает, должны ли временные метки Timestamp быть включены по умолчанию при отправке сообщений (подсказка для JMS‑брокера). Если брокер принимает подсказку, временная метка устанавливается в 0; если игнорирует — используется обычное значение. | True | Логическое значение |
| Pub Sub No Local | pubSubNoLocal | При включении предотвращает получение сообщений, опубликованных тем же соединением. | False | Логическое значение |
| Receive Timeout | receiveTimeout | Время ожидания для получения сообщений (в миллисекундах). | 1 000 | Длинное целое число |
| Recovery Interval | recoveryInterval | Интервал между попытками восстановления (например, при обновлении соединения), в миллисекундах. | 5 000 | Длинное целое число |
| Request Timeout Checker Interval | requestTimeoutCheckerInterval | Задает в миллисекундах, как часто проверять время ожидания запросов/ответов через JMS. Время ожидания определяется параметром Request Timeout. | 1 000 | Длинное целое число |
| Synchronous | synchronous | Определяет, следует ли строго использовать синхронную обработку. | False | Логическое значение |
| Temporary Queue Resolver | temporaryQueueResolver | Позволяет задать собственный распределитель для создания временных очередей. Некоторые системы обмена сообщениями предъявляют особые требования к созданию временных очередей. | Строка | |
| Transfer Exception | transferException | При включении (и использовании обмена InOut) отправляет исключение обратно в ответ в виде jakarta.jms.ObjectMessage, если на стороне потребителя произошел сбой. Если клиент — Camel, возвращенное исключение повторно генерируется. Примечание: Имеет приоритет над TransferExchange. Перехваченное исключение должно быть сериализуемым. Исходное исключение может быть преобразовано во внешнее. | False | Логическое значение |
| Transfer Exchange | transferExchange | Позволяет передавать не только тело сообщения и заголовки, но и все данные объекта Exchange, проходящего через блок JMS (поля: In body, Out body, Fault body, In headers, Out headers, Fault headers, свойства exchange, exchange exception).Требуется, чтобы объекты были сериализуемыми. Несериализуемые объекты исключаются и регистрируются на уровне WARN. Опцию нужно включить на обеих сторонах: отправителя и потребителя. | False | Логическое значение |
| Use Message ID As Correlation ID | useMessageIDAsCorrelationID | Указывает, следует ли всегда использовать JMSMessageID в качестве JMSCorrelationID для входящих сообщений. | False | Логическое значение |
| Wait For Provision Correlation To Be Updated Counter | waitForProvisionCorrelationToBeUpdatedCounter | Количество попыток обновления временного идентификатора корреляции до фактического идентификатора при выполнении запроса/ответа через JMS (при включенном UseMessageIDAsCorrelationID). | 50 | Целое число |
| Wait For Provision Correlation To Be Updated Thread Sleeping Time | waitForProvisionCorrelation ToBeUpdatedThreadSleepingTime | Интервал в миллисекундах для перехода в спящий режим при ожидании обновления предварительного идентификатора корреляции. | 100 | Длинное целое число |
| Wait For Temporary Reply To To Be Updated Counter | waitForTemporaryReplyToToBeUpdatedCounter | Количество попыток ожидания создания и готовности временной очереди replyTo при выполнении запроса/ответа через JMS. | 200 | Целое число |
| Wait For Temporary Reply To To Be Updated Thread Sleeping Time | waitForTemporaryReplyToToBeUpdatedThread SleepingTime | Интервал в миллисекундах для перехода в спящий режим при ожидании готовности временной очереди ответов. | 100 | Длинное целое число |
| Lazy Create Transaction Manager | lazyCreateTransactionManager | При включении создает JmsTransactionManager, если Transacted=True и не задан TransactionManager. | True | Логическое значение |
| Transaction Manager | transactionManager | Используемый диспетчер транзакций. | Строка | |
| Transaction Name | transactionName | Название используемой транзакции. | Строка | |
| Transaction Timeout | transactionTimeout | Тайм‑аут транзакции (в секундах) при использовании режима транзакции. Значение "-1" означает отсутствие ограничений. | -1 | Целое число |
| Disabled | disabled | Доступен только для блоков-отправителей Определяет будет ли отключен блок в потоке. В активированном потоке функциональность блока не будет отрабатываться. | False | Логическое значение |
| Pattern | pattern | Доступен только для блоков-отправителей Шаблон взаимодействия (Exchange Pattern) для обмена сообщениями. Определяет направление потока данных между его компонентами. Возможные значения:
| Список |
Параметры безопасности
| Название в UI | Название атрибута | Описание | Значение по умолчанию | Тип данных |
|---|---|---|---|---|
| Password | password | Пароль для использования в ConnectionFactory. Вы также можете настроить имя пользователя/пароль непосредственно в ConnectionFactory. | Строка | |
| Username | username | Имя пользователя для использования в ConnectionFactory. Вы также можете настроить имя пользователя/пароль непосредственно в ConnectionFactory. | Строка |