2장. ActiveMQ

ActiveMQ 구성 요소를 사용하여 JMS Queue 또는 Topic에 메시지를 보내거나 Apache ActiveMQ 를 사용하여 JMS Queue 또는 Topic에서 메시지를 사용할 수 있습니다.

이 구성 요소는 JMS 구성 요소를 기반으로 하며, Spring의 JmsTemplate을 사용하여 Spring의 JmsTemplate 을 사용하여 MessageListenerContainer 를 사용하여 선언적 트랜잭션에 Spring의 JMS 지원을 사용합니다. 모든 JMS 구성 요소 옵션은 ActiveMQ 구성 요소에도 적용됩니다.

이 구성 요소를 사용하려면 camel-core.jar,camel-spring.jar 및 camel-jms.jar 와 같은 Apache Camel 종속 항목과 함께 classpath에 activemq.jar 또는 activemq-core.jar 가 있는지 확인하십시오.

activemq:[queue:|topic:]destinationName

여기서 destinationName 은 ActiveMQ 큐 또는 주제 이름입니다. 기본적으로 destinationName 은 큐 이름으로 해석됩니다. 예를 들어 큐에 연결하려면 다음을 사용합니다.

activemq:FOO.BAR

원하는 경우 선택적 queue: 접두사를 포함할 수 있습니다.

activemq:queue:FOO.BAR

항목에 연결하려면 topic: 접두사를 포함해야 합니다. 예를 들어, topic에 연결하려면 prices.Prices 를 사용합니다.

activemq:topic:Stocks.Prices

모든 JMS 구성 요소 옵션은 ActiveMQ 구성 요소에도 적용됩니다.

이 구성 요소는 Red Hat JBoss EAP(JBoss Enterprise Application Platform) 컨테이너에 단순화된 배포 모델을 제공하는 EAP의 Camel(Wildfly Camel) 프레임워크에서 지원합니다.

포함된 브로커 또는 외부 브로커에서 작동하도록 ActiveMQ Camel 구성 요소를 구성할 수 있습니다. JBoss EAP 컨테이너에 브로커를 포함하려면 EAP 컨테이너 구성 파일 Cryostat->-<for에서 ActiveMQ 리소스 어댑터를 구성하려면 ActiveMQ 리소스 어댑터 구성 을 참조하십시오.

다음 테스트 사례는 ActiveMQ 연결에 사용되는 brokerURL 을 지정하는 동안 활성MQComponent() 메서드 를 사용하여 CamelContext 에 ActiveMQComponent를 추가하는 방법을 보여줍니다.

camelContext.addComponent("activemq", activeMQComponent("vm://localhost?broker.persistent=false"));

다음과 같이 ActiveMQ 브로커 URL을 ActiveMQComponent에서 구성할 수 있습니다.

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

  <camelContext xmlns="http://camel.apache.org/schema/spring">
  </camelContext>

  <bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
    <property name="brokerURL" value="tcp://somehost:61616"/>
  </bean>

</beans>

Camel을 사용하여 ActiveMQ 브로커로 보낼 때는 풀링된 연결 팩토리를 사용하여 JMS 연결, 세션 및 생산자의 효율적인 풀링을 처리하는 것이 좋습니다. 자세한 내용은 ActiveMQ Spring Support 에서 참조하십시오.

    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-pool</artifactId>
      <version>5.11.0.redhat-630516</version>
    </dependency>

    <bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616" />
    </bean>

    <bean id="pooledConnectionFactory"    class="org.apache.activemq.pool.PooledConnectionFactory" init-method="start" destroy-method="stop">
        <property name="maxConnections" value="8" />
        <property name="connectionFactory" ref="jmsConnectionFactory" />
    </bean>

    <bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
        <property name="connectionFactory" ref="pooledConnectionFactory"/>
        <property name="concurrentConsumers" value="10"/>
    </bean>

    <bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
        <property name="configuration" ref="jmsConfig"/>
    </bean>

그러면 PooledConnectionFactory 가 동시에 사용 중인 최대 8개의 연결이 있는 연결 풀을 생성합니다. 각 연결은 여러 세션에서 공유할 수 있습니다.

연결당 최대 세션 수를 구성하는 데 사용할 수 있는 maxActive 라는 옵션이 있습니다. 기본값은 500 입니다.

ActiveMQ 5.7 이상에서 옵션의 이름이 maxActiveSessionPerConnection 으로 변경되어 더 나은 용도를 반영합니다.

concurrentConsumers 는 maxConnections 보다 높은 값으로 설정됩니다. 이는 각 소비자가 세션을 사용하고 세션에서 동일한 연결을 공유할 수 있으므로 이 작업이 허용됩니다. 이 예제에서는 8 * 500 = 4000 개의 활성 세션을 동시에 가질 수 있습니다.

ActiveMQ 구성 요소는 JMS MessageListener 에서 프로세서 까지 도우미 유형 도 제공합니다. 즉, Cryostat 구성 요소는 모든 경로 내에서 직접 JMS MessageListener Quarkus를 호출할 수 있습니다.

다음과 같이 JMS에서 MessageListener를 생성할 수 있습니다.

public class MyListener implements MessageListener {
   public void onMessage(Message jmsMessage) {
       // ...
   }
}

그런 다음 경로에서 다음과 같이 사용하십시오.

from("file://foo/bar").
  bean(MyListener.class);

즉, Apache Camel 구성 요소를 재사용하여 JMS MessageListener Cryostat\에 쉽게 통합할 수 있습니다!

ActiveMQ 5.6부터 사용 가능

" 대상" 접두사를 사용하여 엔드포인트 uri에서 대상 옵션을 구성할 수 있습니다. 예를 들어 소비자를 독점적으로 표시하고 사전 페치 크기를 50으로 설정하려면 다음과 같이 수행할 수 있습니다.

<camelContext xmlns="http://camel.apache.org/schema/spring">
  <route>
    <from uri="file://src/test/data?noop=true"/>
    <to uri="activemq:queue:foo"/>
  </route>
  <route>
    <!-- use consumer.exclusive ActiveMQ destination option, notice we have to prefix with destination. -->
    <from uri="activemq:foo?destination.consumer.exclusive=true&amp;destination.consumer.prefetchSize=50"/>
    <to uri="mock:results"/>
  </route>
</camelContext>

ActiveMQ는 사용할 수 있는 항목에 배치되는 권고 메시지를 생성할 수 있습니다. 이러한 메시지는 느린 소비자를 감지하거나 통계를 구축하는 경우 경고를 보내는 데 도움이 될 수 있습니다(일당 메시지/생성 횟수 등) 다음 Spring DSL 예제에서는 주제에서 메시지를 읽는 방법을 보여줍니다.

<route>
	<from uri="activemq:topic:ActiveMQ.Advisory.Connection?mapJmsMessage=false" />
	<convertBodyTo type="java.lang.String"/>
	<transform>
	     <simple>${in.body}&#13;</simple>
	</transform>
	<to uri="file://data/activemq/?fileExist=Append&ileName=advisoryConnection-${date:now:yyyyMMdd}.txt" />
</route>

큐에서 메시지를 사용하는 경우 data/activemq 폴더 아래에 다음 파일이 표시됩니다.

advisoryConnection-20100312.txt

advisoryProducer-20100312.txt

포함 된 문자열:

      ActiveMQMessage {commandId = 0, responseRequired = false, messageId = ID:dell-charles-3258-1268399815140
      -1:0:0:0:221, originalDestination = null, originalTransactionId = null, producerId = ID:dell-charles-
      3258-1268399815140-1:0:0:0, destination = topic://ActiveMQ.Advisory.Connection, transactionId = null,
      expiration = 0, timestamp = 0, arrival = 0, brokerInTime = 1268403383468, brokerOutTime = 1268403383468,
      correlationId = null, replyTo = null, persistent = false, type = Advisory, priority = 0, groupID = null,
      groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null,
      marshalledProperties = org.apache.activemq.util.ByteSequence@17e2705, dataStructure = ConnectionInfo
      {commandId = 1, responseRequired = true, connectionId = ID:dell-charles-3258-1268399815140-2:50,
      clientId = ID:dell-charles-3258-1268399815140-14:0, userName = , password = *****,
      brokerPath = null, brokerMasterConnector = false, manageable = true, clientMaster = true},
      redeliveryCounter = 0, size = 0, properties = {originBrokerName=master, originBrokerId=ID:dell-charles-
      3258-1268399815140-0:0, originBrokerURL=vm://master}, readOnlyProperties = true, readOnlyBody = true,
      droppable = false}

이 종속성이 필요합니다.

ActiveMQ는 ActiveMQ 프로젝트와 함께 릴리스된 JMS 구성 요소의 확장입니다.

<dependency>
    <groupId>org.fusesource</groupId>
    <artifactId>camel-activemq</artifactId>
    <version>7.11.0.fuse-sb2-7_11_0-00035-redhat-00001</version>
</dependency>