2장. ActiveMQ

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

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

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

activemq:[queue:|topic:]destinationName

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

activemq:FOO.BAR

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

activemq:queue:FOO.BAR

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

activemq:topic:Stocks.Prices

이 구성 요소에도 적용되는 모든 옵션이 적용되므로 178장. JMS 구성 요소 구성 요소의 옵션을 참조하십시오.

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

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

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

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

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

<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 페이지에 설명되어 있습니다.

Maven으로 Jencks AMQ 풀을 캡처할 수 있습니다.

    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-pool</artifactId>
      <version>5.3.2</version>
    </dependency>

그런 다음 다음과 같이 activemq 구성 요소를 설정합니다.

    <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에서 프로세서 로 도우미 유형 변환기 를 제공합니다. 즉, 43장. 빈 구성 요소 구성 요소는 모든 경로 내에서 직접 JMS MessageListener 빈을 호출할 수 있습니다.

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

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

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

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

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

ActiveMQ 5.6에서 사용 가능

"destination." 접두사를 사용하여 끝점 URI에서 대상 옵션 을 구성할 수 있습니다. 예를 들어 소비자를 배타적으로 표시하고 prefetch 크기를 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 폴더에 다음 파일이 표시됩니다.If you consume a message on a queue, you should see the following files under data/activemq folder:

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 프로젝트 와 함께 릴리스된 178장. JMS 구성 요소 구성 요소의 확장입니다.

<dependency>
  <groupId>org.apache.activemq</groupId>
  <artifactId>activemq-camel</artifactId>
  <version>5.6.0</version>
</dependency>