一、简介
关于jms，请参阅https://docs.oracle.com/javaee/6/tutorial/doc/bncdr.html，看内容是做rpc服务的东西，公司用的是weblogic自带的jms，其实原理差不多，都是创建一个消息工厂，添加队列，从队列中发消息；现在应该不少公司再用dubbo和zookeeper做吧，这个以后再研究。
二、使用
1.添加依赖
<!-- activemq -->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.7.0</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.12.1</version>
</dependency>
<!-- xbean 如<amq:connectionFactory /> -->
<dependency>
<groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId>
<version>3.16</version>
</dependency>
2.生产消费者模式，其实就是发送者和接收者
package com.web.jms.service;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import javax.jms.*;
@Service
public class ProducerService {
@Resource(name = "jmsQueueTemplate")
private JmsTemplate jmsTemplate;
@Resource(name = "queueDestination")
private Destination responseDestination;
public void sendMessage(final String msg){
System.out.println(">>>>\n"+Thread.currentThread().getName()+" 向队列"+responseDestination+"发送消息---------------------->"+msg);
jmsTemplate.send(new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage(msg);
textMessage.setJMSReplyTo(responseDestination);
return textMessage;
}
});
}
}
消费者
package com.web.jms.service;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.TextMessage;
@Service
public class ConsumerService {
@Resource(name = "jmsQueueTemplate")
private JmsTemplate jmsTemplate;
public TextMessage receive(Destination destination){
TextMessage textMessage = (TextMessage) jmsTemplate.receive(destination);
try{
System.out.println("从队列" + destination.toString() + "收到了消息：\t"
+ textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
return textMessage;
}
}
3.消费者监听
package com.web.jms.listener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class JmsListener implements MessageListener {
public void onMessage(Message message) {
TextMessage textMsg = (TextMessage) message;
try {
System.out.println("接收到了消息，消息内容是：" + textMsg.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
4.errorhandler
package com.web.jms.handler;
import org.springframework.stereotype.Service;
import org.springframework.util.ErrorHandler;
@Service
public class JmsErrorHandler implements ErrorHandler{
public void handleError(Throwable throwable) {
System.out.println("出错了！");
throwable.printStackTrace();
}
}
5.配置applicationContext-ActiveMQ.xml
<?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:context="http://www.springframework.org/schema/context"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:amq="http://activemq.apache.org/schema/core"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:mvc="http://www.alibaba.com/schema/stat"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop-4.1.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-4.1.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.1.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core-5.7.0.xsd
http://www.alibaba.com/schema/stat http://www.alibaba.com/schema/stat.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms-4.1.xsd ">
<context:annotation-config />
<!-- ActiveMQ 连接工厂 -->
<!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->
<!-- 如果连接网络：tcp://ip:61616；未连接网络：tcp://localhost:61616 以及用户名，密码-->
<amq:connectionFactory id="amqConnectionFactory"
brokerURL="tcp://localhost:61616"
userName="admin" password="admin" />
<!-- Spring Caching连接工厂 -->
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
<!-- 同上，同理 -->
<!-- <constructor-arg ref="amqConnectionFactory" /> -->
<!-- Session缓存数量 -->
<property name="sessionCacheSize" value="100" />
</bean>
<!-- Spring JmsTemplate 的消息生产者 start-->
<!-- 定义JmsTemplate的Queue类型 -->
<bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<constructor-arg ref="connectionFactory" />
<!-- 非pub/sub模型（发布/订阅），即队列模式 -->
<property name="pubSubDomain" value="false" />
<property name="defaultDestination" ref="queueDestination" />
</bean>
<!--Spring JmsTemplate 的消息生产者 end-->
<!-- 消息消费者 start-->
<!-- 配置消息队列监听者（Queue） -->
<bean id="queueMessageListener" class="com.web.jms.listener.JmsListener" />
<!-- 显示注入消息监听容器（Queue），配置连接工厂，监听的目标是Destination，监听器是上面定义的监听器 -->
<bean id="queueListenerContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="errorHandler" ref="JmsErrorHandler"/>
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="queueDestination" />
<property name="messageListener" ref="queueMessageListener" />
</bean>
<bean id="JmsErrorHandler" class="com.web.jms.handler.JmsErrorHandler"></bean>
<!-- 消息消费者 end -->
<!-- 定义队列目的地 ，点对点 -->
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg>
<value>queue</value>
</constructor-arg>
</bean>
</beans>
6.web.xml中添加启动文件
<!-- 1.针对Spring配置：读取配置文件 -->
<context-param>
<param-name>contextConfigLocation</param-name>
<param-value>
classpath:config/applicationContext.xml,
classpath:config/applicationContext-ActiveMQ.xml
</param-value>
</context-param>
7.controller
package com.web.jms.controller;
import com.web.jms.service.ConsumerService;
import com.web.jms.service.ProducerService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;
import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.TextMessage;
import java.text.SimpleDateFormat;
import java.util.Date;
@Controller
@RequestMapping("/jms")
public class JmsController {
@Resource(name = "queueDestination")
private Destination destination;
//队列消息生产者
@Autowired
private ProducerService producer;
//队列消息消费者
@Autowired
private ConsumerService consumer;
@RequestMapping(value = "/SendMessage", method = RequestMethod.GET)
@ResponseBody
public void send() {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:ss");
String msg = sdf.format(new Date());
System.out.println(Thread.currentThread().getName()+"------------send to jms Start");
producer.sendMessage(msg);
System.out.println(Thread.currentThread().getName()+"------------send to jms End");
}
@RequestMapping(value= "/ReceiveMessage",method = RequestMethod.GET)
@ResponseBody
public void receive(){
System.out.println(Thread.currentThread().getName()+"------------receive from jms Start");
TextMessage tm = consumer.receive(destination);
System.out.println(Thread.currentThread().getName()+"------------receive from jms End");
}
}
-end-
网友评论