activemq spring整合 以及虚拟主题设置

1. jmstemplate 配置

<bean id="jmstemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="mqConnectionFactory" />
<property name="defaultDestination" ref="testqueue"/>
<property name="deliveryMode" value="1"></property>
<property name="sessionTransacted" value="false"/>
<!-- deliveryMode, priority, timeToLive 的开关，要生效，必须配置为true，默认false-->
<property name="explicitQosEnabled" value="false" />
<!-- 超时时间 -->
<property name="timeToLive" value="${acq.query.timeout}"/>
</bean>

class="org.apache.activemq.command.ActiveMQQueue"><constructor-arg value="com.cn.testqueue" /></bean>

<bean id="mqConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="${mq.brokerURL}" />
</bean>
</property>
</bean>

<p><em><span style="color:#2A00FF;">MQ_URL=failover:(tcp://ip:61616,tcp://ip:</span> <span style="color:#2A00FF;">61616)</span></em></p>

2.消息监听器配置

<!-- 消息监听器 -->
<bean id="consumerMessageListener" class="com.activemq.lisenter.MessageListenterA"/>
<bean id="jmsContainer"        class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="mqConnectionFactory">
<property name="destination" ref="testqueue">
<property name="messageListener" ref="consumerMessageListener" />
<property name="sessionTransacted" value="true"></property>
<property name="concurrentConsumers" value="2"/>
</bean>

3.引入 jar 包

<properties>
<activemq.version>5.9.1</activemq.version>
</properties>

<!-- activemq -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-broker</artifactId>
<version>${activemq.version}</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
<version>${activemq.version}</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-console</artifactId>
<version>${activemq.version}</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-jaas</artifactId>
<version>${activemq.version}</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-openwire-legacy</artifactId>
<version>${activemq.version}</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-spring</artifactId>
<version>${activemq.version}</version>
</dependency>

4. 消息发送

jmsTemplate.convertAndSend(Object message);

5. 消息异步接收

实现监听器MessageListener接口，配置成Spring Bean。只有一个方法需实现：
public void onMessage(Message message);

例如：
public class <span style="font-family: Arial, Helvetica, sans-serif;">MessageListenterA </span><span style="font-family: Arial, Helvetica, sans-serif;">implements MessageListener {</span>

@Override
public void onMessage(Message message) {
if (!(message instanceof ObjectMessage)) {
logger.info("接收的消息类型不匹配！");
return;
}

try {
ObjectMessage objectMessage = (ObjectMessage) message;
ShortMessage smsMsg = (ShortMessage)objectMessage.getObject();

//…………………

} catch (JMSException e) {
logger.error("jms异常", e);
}

}

注意：activemq 使用PooledConnectionFactory连接池引入 commons-pool2，如果weblogic版本较低 会与weblogic 的jar 包冲突，此时需要调整weblogic 加载顺序

在 web-inf 目录下面加入 weblogic.xml

<?xml version="1.0" encoding="UTF-8"?>
<weblogic-web-app
xmlns="http://xmlns.oracle.com/weblogic/weblogic-web-app"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/ejb-jar_3_0.xsd http://xmlns.oracle.com/weblogic/weblogic-web-app http://xmlns.oracle.com/weblogic/weblogic-web-app/1.2/weblogic-web-app.xsd"> 
<container-descriptor>
<!--<prefer-web-inf-classes>true</prefer-web-inf-classes>-->
<prefer-application-packages>
<package-name>org.apache.commons.pool*</package-name>
</prefer-application-packages>
</container-descriptor>
</weblogic-web-app>

activemq虚拟主题 是为了解决同一个主题多个应用订阅，同一个应用集群的多台机器订阅只消费一次 ，主题订阅 注意 命名 是VirtualTopic开头，主题必须持久化，同一个应用clientId 相同

<!--主题订阅-->
<bean id="queue_market_topic" class="org.apache.activemq.command.ActiveMQTopic">
<!--对列名称自定义，如果Activemq没有这个对列会新建-->
<constructor-arg value="VirtualTopic.TEST" />
</bean>

    <bean id="myListenerContainer"  
        class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
        <property name="connectionFactory" ref="connectionFactory" />  
        <property name="pubSubDomain" value="true"/>  
        <property name="subscriptionDurable" value="true"/>  
        <property name="receiveTimeout" value="10000"/>  
        <property name="clientId" value="push_client_119" />  
        <property name="durableSubscriptionName" value="push_client_119"/>  
        <property name="destination" ref="myDestination" />  
        <property name="messageListener" ref="myTopicListenerA" />  
    </bean>