spring集成mq相关

1.spring集成mq的步骤(客户端)

<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->  

<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">  

  <property name="brokerURL" value="tcp://localhost:61616"/>  

</bean>  

<!-activemq提供的一个连接池,减少资源消耗,也可以不用->

<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">  

<property name="connectionFactory" ref="targetConnectionFactory"/>  

<property name="maxConnections" value="10"/>  

</bean>  

<!--spring提供的集成mq的接口,把厂商提供的mq连接工厂注入进来。-->

<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">  

<property name="targetConnectionFactory" ref="pooledConnectionFactory"/>  

</bean>  

<!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 ,把spring的连接工厂注入进来,用户消息的收发-->  

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  

  <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  

<property name="connectionFactory" ref="connectionFactory"/>  

</bean>  

<!--这个是队列目的地,点对点的-->  

<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  

    <constructor-arg>  

<value>queue</value>  

</constructor-arg>  

</bean>  

<!--这个是主题目的地,一对多的-->  

<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">  

<constructor-arg value="topic"/>  

</bean

<!--通过spring提供的方法,连接工厂已经注入的方法,来进行发送消息,发送的方向为destination,这个仅仅是个名字。在spring里面定义好的--> 

jmsTemplate.send(destination, new MessageCreator() {  

public Message createMessage(Session session) throws JMSException {  

return session.createTextMessage(message);  

    }  

    });

2.mq的服务端

mq的服务端实现对mq客户端发来的信息进行实施的监控。主要的配置为:

<!--mq的连接工厂-->

<bean id="mqConnectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
<property name="brokerURL" value="${xcm.mqQueue.brokerURL}" />
<property name="userName" value="${xcm.mqQueue.userName}" />
<property name="password" value="${xcm.mqQueue.password}" />
</bean>

<!--mq的名称,也就是队列名称-->

<bean id="mqQueue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="${xcm.fdMqQueue.name}" />
</bean>


<!--mq的监听实现类-->
<bean id="fundMqListener" class="com.xcm.fund.core.service.impl.mq.FundMqListener">
</bean>


<!--mq的监听容器。-->
<bean id="jmsContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="mqConnectionFactory" />
<property name="destination" ref="mqQueue" />
<property name="messageListener" ref="fundMqListener" />
<property name="concurrentConsumers" value="2"/>
</bean>

牵扯到事务的配置如下:

<bean id="jmsContainer"  

 class="org.springframework.jms.listener.DefaultMessageListenerContainer"

<property name="connectionFactory" ref="connectionFactory" />  

<property name="destination" ref="queueDestination" />  

<property name="messageListener" ref="consumerMessageListener" />  

<property name="sessionTransacted" value="true"/>  (开启事务,如果接收的时候报异常,则消息回滚,此时下次接收还会收到异常的那个信息)

</bean>  

使用为:

public class ConsumerMessageListener implements MessageListener {     

public void onMessage(Message message) {  

//这里我们知道生产者发送的就是一个纯文本消息,所以这里可以直接进行强制转换,或者直接把onMessage方法的参数改成Message的子类TextMessage  

            TextMessage textMsg = (TextMessage) message;  

            System.out.println("接收到一个纯文本消息。");  

try {  

                System.out.println("消息内容是:" + textMsg.getText());  

     if (1 == 1) {  

                throw new RuntimeException("Error");  (抛出异常,下次还会接收到)

              }  

            } catch (JMSException e) {  

                e.printStackTrace();  

           }  

    }  

}

和数据库在同一个事务性的配置如下:

<bean id="jmsContainer"  

 class="org.springframework.jms.listener.DefaultMessageListenerContainer">  

<property name="connectionFactory" ref="connectionFactory" />  

<property name="destination" ref="queueDestination" />  

<property name="messageListener" ref="consumerMessageListener" />  

<property name="transactionManager" ref="jtaTransactionManager"/>  

</bean>  

<bean id="jtaTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager"/>  

数据库配置:

<bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">  

 <property name="dataSource" ref="dataSource"/>  

</bean>  

<jee:jndi-lookup jndi-name="jdbc/mysql" id="dataSource"/>  

<bean id="jtaTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager"/>  

<tx:annotation-driven transaction-manager="jtaTransactionManager"/> 

原文地址:https://www.cnblogs.com/yanghuiping/p/9288584.html