active mq 消费

首先配置MQ connection

<bean id="amqConnectionFactory"
class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL"
value="failover://@{amq.address}?randomize=false&jms.useAsyncSend=true&jms.prefetchPolicy.queuePrefetch=1" />
</bean>
<bean id="pollConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" >
<property name="connectionFactory" ref="amqConnectionFactory" />
<property name="maxConnections"  value="@{maxConnections}"/>
</bean>

配置队列

<bean id="priceInfoQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="@{price.amq.queue.name}" />
</bean>

配置listener

<bean id="priceListener" class="com.mq.listener.ProductPriceListener" />
<bean id="priceConsume"
class="com.mq.consume.PriceMQConsume">
<property name="connectionFactory" ref="amqConnectionFactory" />
<property name="destination" ref="priceInfoQueue" />
<property name="messageListener" ref="priceListener"  />
<property name="receiveTimeout" value="10000" />
<property name="sessionTransacted" value="true" />
<property name="sessionAcknowledgeMode" value="0" />
</bean>

监听类 实现messageListener

public class ProductPriceListener

implements MessageListener { private static final Logger logger = LoggerFactory.getLogger(

ProductPriceListener

.class); @Override public void onMessage(Message message) { if (!(message instanceof TextMessage)) { logger.warn("Only TextMessage will be processed."); return; } try {//消费代码 } catch (Exception e){ //异常 } }}

消费容器可以使用

org.springframework.jms.listener.DefaultMessageListenerContainer

public class PriceMQConsume extends DefaultMessageListenerContainer {
}