一、创建配置消息发送接收目的地、 ActiveMQ中间件地址
JMS_BROKER_URL=failover://(tcp://192.168.1.231:61616) QUEUE_BUSP_TP_SMS_MESSAGE=busp.tp.sms.message
二、创建消息生产者配置
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:jdbc="http://www.springframework.org/schema/jdbc" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:jee="http://www.springframework.org/schema/jee" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.1.xsd http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-3.1.xsd http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-2.5.xsd"> <!-- load config file --> <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> <property name="ignoreUnresolvablePlaceholders" value="true" /> <property name="locations"> <list> <value>classpath:property/sms-jms.properties</value> </list> </property> </bean> <!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供--> <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="${JMS_BROKER_URL}"/> </bean> <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory --> <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory --> <property name="targetConnectionFactory" ref="targetConnectionFactory"/> </bean>
<!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 --> <property name="connectionFactory" ref="connectionFactory"/> </bean>
<!-- 发送资金流水-->
<bean id="sendMessageDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>${QUEUE_BUSP_TP_SMS_MESSAGE}</value>
</constructor-arg>
</bean>
</beans>
三、创建生产者并发送消息
package com.busp.tp.sms.sender; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session; import org.apache.log4j.Logger; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; import com.busp.tp.sms.common.SMSMsg; /** * @ClassName: PushJMSSender * @version 1.0 * @Desc: 向jms发送消息 * @date 2016年4月21日下午6:01:26 * @history v1.0 * */ @SuppressWarnings("resource") public class SMSJMSSender { // 日志文件 private static Logger logger = Logger.getLogger(SMSJMSSender.class); // jms template private static JmsTemplate jmsTemplate; // 目标地址 private static Destination sendMessageDestination; static{ ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:sms-jms-sender.xml"); ctx.registerShutdownHook(); sendMessageDestination = (Destination) ctx.getBean("sendMessageDestination"); jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate"); } /** * * 描述:向JMS推送消息(批量) * @date 2016年4月21日下午6:31:48 * @param pushMsg */ public static void sendJMSMessage(SMSMsg smsMsg){ try { final String result = smsMsg.toJSONString(); logger.info("向JMS推送消息:" + result); jmsTemplate.send(sendMessageDestination, new MessageCreator() { public Message createMessage(Session session) throws JMSException { return session.createTextMessage(result); } }); } catch (Exception e) { logger.error("向JMS推送消息发生了异常",e); } } }
四、创建接收者配置
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd" default-lazy-init="false"> <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> <property name="ignoreUnresolvablePlaceholders" value="true" /> <property name="locations"> <list> <value>classpath:property/sms-jms.properties</value> </list> </property> </bean> <!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供--> <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="${JMS_BROKER_URL}"/> </bean> <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory --> <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory --> <property name="targetConnectionFactory" ref="targetConnectionFactory"/> </bean> <!-- ################################### 监听放款信息 Start #####################################--> <bean id="smsMessageQueueDestination" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg> <value>${QUEUE_BUSP_TP_SMS_MESSAGE}</value> </constructor-arg> </bean> <bean id="smsMessageListener" class="com.busp.tp.sms.listener.SMSMessagelListener"/> <bean id="smsMessageContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="targetConnectionFactory" /> <property name="destination" ref="smsMessageQueueDestination" /> <property name="messageListener" ref="smsMessageListener" /> </bean> <!-- ################################### 监听放款信息 End ############################# </beans>
五、接受者接收消息
package com.busp.tp.sms.listener; import javax.annotation.Resource; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; import org.apache.log4j.Logger; import com.busp.common.util.string.StringUtil; import com.busp.tp.sms.handler.SMSHandler; /** * @ClassName: PushMessageListener * @version 1.0 * @Desc: TODO * @date 2016年4月21日下午4:19:18 * @history v1.0 * */ public class SMSMessagelListener implements MessageListener{ // 日志文件 private Logger logger = Logger.getLogger(SMSMessagelListener.class); @Resource private SMSHandler yxtSMSHandler; //@Resource //private PushMsgHandler androidPushMsgHandler; @Override public void onMessage(Message message) { try { TextMessage text = (TextMessage) message; final String value = text.getText(); logger.info("接收到客户端发来的消息:"+ value); } catch (JMSException e) { logger.error("",e); } } }