Интеграция с RabbitMQ

Описание

В данном примере демонстрируется настройка интеграции с RabbitMQ в Entaxy ION: создание входного и выходного коннекторов, отправка и получение сообщений, а также обработка ошибок с использованием механизма NACK.

Процесс взаимодействия с RabbitMQ состоит из следующих шагов:

  1. Запустить RabbitMQ в докере:

    docker pull rabbitmq:4.0.9-management
docker run --name rabbitmq -p 5673:5672 -p 15672:15672 rabbitmq:4.0.9-management

  2. Установить фичу camel-rabbitmq:

    feature:install -r camel-rabbitmq

  3. Убедиться, что установленные бандлы находятся в статусе Active:

    695 │ Active    │  50 │ 5.9.0              │ RabbitMQ Java Client
696 │ Active    │  50 │ 4.1.11             │ Metrics Core
697 │ Active    │  50 │ 3.4.5              │ camel-rabbitmq

  4. Создать профиль системы с именем rabbit и добавить к нему:

    1. Входной кастомный коннектор (consumer). В точке кастомизации custom-route создаем маршрут принимающий сообщения из очереди:

      <?xml version="1.0" encoding="UTF-8"?>
<entaxy:object-input-route
	xmlns="http://camel.apache.org/schema/blueprint"
	xmlns:blueprint="http://www.osgi.org/xmlns/blueprint/v1.0.0"
	xmlns:entaxy="http://www.entaxy.ru/schemas/1.0"
	xmlns:m="http://www.entaxy.ru/schemas/entaxy-mediators/1.0">
	<blueprint:bean id="rabbitConnectionFactory" class="com.rabbitmq.client.ConnectionFactory">
		<blueprint:property name="host" value="localhost"/>
		<blueprint:property name="port" value="5673"/>
		<blueprint:property name="username" value="guest"/>
		<blueprint:property name="password" value="guest"/>
	</blueprint:bean>
	<from uri="rabbitmq:exrabbit?prefetchEnabled=true&amp;prefetchCount=1&amp;queue=exQueue"/>
	<m:log message="In Rabbit consumer: ${body}"/>
	<m:respond now="true" continue="false"/>
</entaxy:object-input-route>

    2. Выходной кастомный коннектор (producer). В точке кастомизации custom-route создаем маршрут отправляющий сообщения в очередь:

      <?xml version="1.0" encoding="UTF-8"?>
<entaxy:object-route
    xmlns="http://camel.apache.org/schema/blueprint"
    xmlns:blueprint="http://www.osgi.org/xmlns/blueprint/v1.0.0"
    xmlns:entaxy="http://www.entaxy.ru/schemas/1.0"
    xmlns:m="http://www.entaxy.ru/schemas/entaxy-mediators/1.0">
    <blueprint:bean id="rabbitConnectionFactory" class="com.rabbitmq.client.ConnectionFactory">
        <blueprint:property name="host" value="localhost"/>
        <blueprint:property name="port" value="5673"/>
        <blueprint:property name="username" value="guest"/>
        <blueprint:property name="password" value="guest"/>
    </blueprint:bean>
    <setBody>
        <constant>To Local Rabbit</constant>
    </setBody>
    <m:log message="Rabbit producer started: ${body}"/>
    <to uri="rabbitmq:exrabbit?queue=exQueue"/>
</entaxy:object-route>

    3. К выходному кастомному коннектору добавляем маршрут на базе компонента таймер, инициирующий вызов коннектора с заданным интервалом:

      <?xml version="1.0" encoding="UTF-8"?>
<entaxy:common-route id="common-route-2250"
    xmlns="http://camel.apache.org/schema/blueprint"
    xmlns:blueprint="http://www.osgi.org/xmlns/blueprint/v1.0.0"
    xmlns:entaxy="http://www.entaxy.ru/schemas/1.0"
    xmlns:m="http://www.entaxy.ru/schemas/entaxy-mediators/1.0">
    <m:log message="in route ${routeId}" loggingLevel="INFO"/>
    <to uri="system:rabbit?connectorClassifier=main"/>
</entaxy:common-route>

  5. Проверка работы

    1. Перейти в веб-консоль RabbitMQ (http://localhost:15672)

    2. Отправить тестовое сообщение в очередь exQueue и убедиться, что оно получено и отображается в логах:

      16:32:24.560 INFO [Camel (rabbit.custom-connector-in.main) thread #132 - rabbitmq://extest] d1553b08-d035-404b-b2c6-362212064062#{"service":"rabbit.custom-connector-in.main","sender":"rabbit"}# In Rabbit consumer: TEST

  6. Обработка ошибок и использование NACK

    Для управления повторной обработки сообщений используется механизм NACK (Negative Acknowledgement). В данном примере NACK реализуется в маршруте входного кастомного коннектора:

    <?xml version="1.0" encoding="UTF-8"?>
<entaxy:object-input-route
    xmlns="http://camel.apache.org/schema/blueprint"
    xmlns:blueprint="http://www.osgi.org/xmlns/blueprint/v1.0.0"
    xmlns:entaxy="http://www.entaxy.ru/schemas/1.0"
    xmlns:m="http://www.entaxy.ru/schemas/entaxy-mediators/1.0">
    <blueprint:bean id="rabbitConnectionFactory" class="com.rabbitmq.client.ConnectionFactory">
        <blueprint:property name="host" value="localhost"/>
        <blueprint:property name="port" value="5673"/>
        <blueprint:property name="username" value="guest"/>
        <blueprint:property name="password" value="guest"/>
    </blueprint:bean>
    <onException>
        <exception>java.lang.Exception</exception>
        <handled>
            <constant>true</constant>
        </handled>
        <m:log message="Failed to process message, sending NACK: ${body}"/>
        <!-- NACK the message and requeue it (requeue=true) -->
        <setHeader name="rabbitmq.REQUEUE">
            <constant>true</constant>
        </setHeader>
        <to uri="rabbitmq:nack"/>
    </onException>
    <from uri="rabbitmq:exnack?queue=exNackQueue&amp;autoAck=false"/>
    <m:log message="In Rabbit consumer: ${body}"/>
    <!-- Simulate processing failure -->
    <throwException exceptionType="java.lang.Exception" message="Process Failed"/>
    <m:respond now="true" continue="false"/>
</entaxy:object-input-route>

  7. Проверка работы

    1. Перейти в веб-консоль RabbitMQ (http://localhost:15672)

    2. Отправить тестовое сообщение в очередь exNackQueue и убедиться, что сообщение возвращается в очередь при ошибке:

      16:26:04.236 INFO [Camel (rabbit.custom-connector-in.main) thread #114 - rabbitmq://exnack] c5af3e50-7ef1-4f30-b24a-2d36cd90f775#{"service":"rabbit.custom-connector-in.main","sender":"rabbit"}# In Rabbit consumer: test
16:26:04.237 INFO [Camel (rabbit.custom-connector-in.main) thread #114 - rabbitmq://exnack] c5af3e50-7ef1-4f30-b24a-2d36cd90f775#{"service":"rabbit.custom-connector-in.main","sender":"rabbit"}# Failed to process message, sending NACK: test