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

Описание

В данном примере демонстрируется взаимодействие с брокером сообщений ActiveMQ Classic по протоколу AMQP с использованием Apache Camel. Реализуется отправка и чтение сообщений из очереди.

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

    docker pull apache/activemq-classic:latest
docker run -p 61617:61616 -p 8162:8161 -p 5673:5672 -p 1884:1883 apache/activemq-classic:latest

  2. Создать профиль системы с именем erp и добавить к нему следующие коннекторы:

    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="amqpConnectionFactory" class="org.apache.qpid.jms.JmsConnectionFactory">
        <blueprint:property name="remoteURI" value="amqp://localhost:5673"/>
    </blueprint:bean>
    <blueprint:bean id="amqp" class="org.apache.camel.component.amqp.AMQPComponent">
        <blueprint:property name="connectionFactory" ref="amqpConnectionFactory"/>
    </blueprint:bean>
    <from uri="amqp:queue:test1.queue"/>
    <m:log message="Custom ActiveMQ classic input route start"/>
    <m:log message="Received: ${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="amqpConnectionFactory" class="org.apache.qpid.jms.JmsConnectionFactory">
        <blueprint:property name="remoteURI" value="amqp://localhost:5673"/>
    </blueprint:bean>
    <blueprint:bean id="amqp" class="org.apache.camel.component.amqp.AMQPComponent">
        <blueprint:property name="connectionFactory" ref="amqpConnectionFactory"/>
    </blueprint:bean>
    <m:log message="Custom output route start"/>
    <transform>
        <simple>Hello World From Custom Connector</simple>
    </transform>
    <to uri="amqp:queue:test1.queue"/>
</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:erp"/>
</entaxy:common-route>

  3. В результате мы получаем:

    • Таймер инициирует вызов выходного коннектора профиля erp

    • В выходном коннекторе (producer) формируется сообщение и отправляется в очередь test1.queue

    • Входной коннектор (consumer) подписан на данную очередь и автоматически получает сообщения

    • Полученные сообщения отображаются в логах:

    17:54:04.735 INFO [Camel (activeMQ.custom-connector-out.main) thread #161 - timer://timer-01]  in route activeMQ.custom-connector-out.main__timer-01
17:54:04.735 INFO [Thread-3765]  Custom output route start
17:54:04.745 INFO [AmqpProvider :(206):[amqp://localhost:5673]] Connection ID:03fa9495-182e-476f-9e2c-40a6b315afc1:205 connected to server: amqp://localhost:5673
17:54:04.761 INFO [Camel (activeMQ.custom-connector-in.main) thread #152 - JmsConsumer[test1.queue]] c3e24840-aa54-4ebb-b438-d67c562bb80e#{"service":"sys-26","sender":"activeMQ"}# Custom ActiveMQ classic input route start
17:54:04.762 INFO [Camel (activeMQ.custom-connector-in.main) thread #152 - JmsConsumer[test1.queue]] c3e24840-aa54-4ebb-b438-d67c562bb80e#{"service":"sys-26","sender":"activeMQ"}# Received: Hello World From Custom Connector
17:54:05.736 INFO [Camel (activeMQ.custom-connector-out.main) thread #161 - timer://timer-01]  in route activeMQ.custom-connector-out.main__timer-01
17:54:05.736 INFO [Thread-3766]  Custom output route start
17:54:05.768 INFO [AmqpProvider :(207):[amqp://localhost:5673]] Connection ID:4b2c78a9-5993-482a-ad6f-17e4dee59367:206 connected to server: amqp://localhost:5673