首页 > 代码库 > Spring整合activeMQ消息队列
Spring整合activeMQ消息队列
1.配置JMS
<!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 --> <property name="connectionFactory" ref="connectionFactory"/> </bean> <!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供--> <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://localhost:61616"/> </bean> <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory --> <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory --> <property name="targetConnectionFactory" ref="targetConnectionFactory"/> </bean>
发送信息到activeMQ
@Override public void addNotifyCashToMq(final String notifyUrl, final String cashId, final String reqSn, final String callResult,int count) { //发送的参数final String callBackUrl = SuperAppConstant.TRANSACTION_CALLBACK_PREFIX_URL + notify_url_notifyCash + notifyUrl + "&cashId=" + cashId + "&reqSn=" + reqSn + "&callResult=" + callResult + "&count=" + _count;
//发送消息到queue_notifuCash_serial消息队列 jmsTemplate.send(queue_notifyCash_serial, new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { if (logger.isDebugEnabled()) { logger.debug("notifyUrl=" + notifyUrl + ",cashId=" + cashId + ",reqSn=" + reqSn + ",callResult=" + callResult + ",_count=" + _count); } HashMap map = new HashMap(); map.put("callBackUrl", callBackUrl); ObjectMessage objectMessage = session.createObjectMessage();//创建消息 objectMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);//延时 return objectMessage; } }); }
xml配置信息
<!-- ActiveMQ 连接工厂 --> <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="http://www.mamicode.com/${jms.broker_url}" /> </bean> <!-- Spring Caching 连接工厂 --> <bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <property name="targetConnectionFactory" ref="connectionFactory" /> <property name="sessionCacheSize" value="http://www.mamicode.com/10" /> </bean> <!-- Spring JMS Template --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="cachingConnectionFactory" /> </bean>
2.destination消息队列定义
<description>Queue定义</description> <bean id="queue_callback_serial" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg> <value>queue_callback_serial</value> </constructor-arg> </bean>
3。监听器BatchJob
3.1 jms.xml
<description>JMS简单应用配置</description> <!-- ActiveMQ 连接工厂 --> <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="http://www.mamicode.com/${jms.broker_url}" /> </bean> <!-- Spring Caching 连接工厂 --> <bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <property name="targetConnectionFactory" ref="connectionFactory" /> <property name="sessionCacheSize" value="http://www.mamicode.com/10" /> </bean> <!-- Queue定义 --> <bean id="orderQueueProducer" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg value="http://www.mamicode.com/order.queue.producer" /> </bean> <!-- Spring JMS Template --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="cachingConnectionFactory" /> <property name="defaultDestination" ref="orderQueueProducer" /> </bean> <!-- 使用Spring JmsTemplate的消息生产者 --> <bean id="orderProducerJmsService" class="com.gmall88.server.jms.order.impl.OrderProducerJmsServiceImpl"> <property name="jmsTemplate" ref="jmsTemplate" /> </bean> <!-- 定义消息队列 --> <bean id="orderQueueListener" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg> <value>order.queue.listener</value> </constructor-arg> </bean>
3.2 监听器impl
import java.util.Map; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.ObjectMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.gmall88.server.wxpay.RF; import net.sf.json.JSONObject; public class NotifyCashManagerImpl implements MessageListener { private Logger logger = LoggerFactory.getLogger(getClass()); @Override public void onMessage(Message message) { if(logger.isDebugEnabled()){ logger.debug("new callback start.."); } if(message !=null){ if(message instanceof ObjectMessage){ ObjectMessage objectMessage = (ObjectMessage) message;//监听消息 try { Map param = (Map)objectMessage.getObject(); String callBackUrl = (String)param.get("callBackUrl");//取出消息里的参数 if (logger.isInfoEnabled()) { logger.info("callBackUrl=" + callBackUrl); } JSONObject jsonObject = RF.httpsRequestJson(callBackUrl, "POST", "");//通过http回调方法 if(jsonObject != null){ logger.info("code:"+jsonObject.getString("code")); logger.info("message="+jsonObject.getString("message")); } } catch (Exception e) { logger.error(e.getMessage(),e); } }else{ logger.error("Unknown message, type=" + message.getClass().getName()); } }else{ logger.error("message is null"); } } }
回调方法:
@RequestMapping(value = "http://www.mamicode.com/notifyCash", method = RequestMethod.POST) @ResponseBody public Object notifyCash(String notifyUrl, String cashId, String reqSn, String cashResult,int count) { ReturnResult returnResult = new ReturnResult(); String clientId = "superApp_notifyOrder"; try { clientId += cashId; returnResult = recordRequestCheck(clientId); if(returnResult != null){ return returnResult; } returnResult = new ReturnResult(); try{ // 回调业务系统 try { superAppServerManager.notifyCash(notifyUrl, cashId, reqSn, cashResult); } catch (Exception e) { // 回调失败,做延时回调 logger.error(e.getMessage(), e); superAppServerManager.addNotifyCashToMq(notifyUrl, cashId, reqSn, cashResult, count); } }finally{ recordRequestEnd(clientId); } } catch (GmallException e) { returnResult.setCodeNum(e.getCode()); returnResult.setMessage(e.getMessage()); } catch (Exception e) { logger.error(e.getMessage(), e); returnResult.setCode(ReturnCodeType.FAILURE) .setMessage(e.getMessage()); } logger.info("called.."); return returnResult; }
整理了一下整个流程如图所示:
Spring整合activeMQ消息队列
声明:以上内容来自用户投稿及互联网公开渠道收集整理发布,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任,若内容有误或涉及侵权可进行投诉: 投诉/举报 工作人员会在5个工作日内联系你,一经查实,本站将立刻删除涉嫌侵权内容。