SSM之集成activeMQ
文章目录
1.介绍
主要实现spring集成activeMQ
2.配置
2.1 添加依赖pom
<!--添加MQ所需Jar-->
<!-- active-mq -->
<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.11.1</version>
</dependency>
<!-- sping 对消息队列的支持 -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>4.3.7.RELEASE</version>
</dependency>
2.2 创建MQ配置文件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:context="http://www.springframework.org/schema/context" xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
<!--1.ActiveMQ 真正产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供 也可以用amq:内置标签来定义-->
<!--<amq:connectionFactory id="amqConnectionFactory"-->
<!--brokerURL="tcp://192.168.1.1:61616"-->
<!--userName="admin" password="admin"></amq:connectionFactory>-->
<bean id="activeMqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
<property name="userName" value="admin"></property>
<property name="password" value="admin"></property>
</bean>
<!--2.配置spring 管理 connectionFactory类-->
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<constructor-arg ref="activeMqConnectionFactory" />
<property name="sessionCacheSize" value="100" />
</bean>
<!--3.配置消息队列生产者与消费者以及监听者-->
<!-- 定义消息队列(Queue) -->
<bean id="myQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<!-- 设置消息队列的名字 -->
<constructor-arg>
<value>MyQueueTest01</value>
</constructor-arg>
</bean>
<!-- 配置JMS模板(Queue),Spring提供的JMS工具类,它发送、接收消息。 -->
<bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="defaultDestination" ref="myQueueDestination" />
<property name="receiveTimeout" value="10000" />
<!-- true是topic,false是queue,默认是false,此处显示写出false -->
<property name="pubSubDomain" value="false" />
</bean>
<!-- 配置消息队列监听者(Queue) -->
<bean id="queueMessageListener" class="com.zy.mq.service.QueueMessageListener" />
<!-- 显示注入消息监听容器(Queue),配置连接工厂,监听的目标是demoQueueDestination,监听器是上面定义的监听器 -->
<bean id="queueListenerContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="myQueueDestination" />
<property name="messageListener" ref="queueMessageListener" />
</bean>
<!--4.配置主题消息生产者以及监听者-->
<!--这个是主题(topic)目的地,一对多的 -->
<bean id="myTopicDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="MyTopic01" />
</bean>
<!-- 配置JMS模板(Topic),Spring提供的JMS工具类,它发送、接收消息。 -->
<bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="defaultDestination" ref="myTopicDestination" />
<property name="receiveTimeout" value="10000" />
<!-- true是topic,false是queue,默认是false,此处显示写出false -->
<property name="pubSubDomain" value="true" />
<!-- 消息不持久化 -->
<property name="explicitQosEnabled" value="false"></property>
</bean>
<!-- 消息监听类 -->
<bean id="topicMessageListener1" class="com.zy.mq.service.TopicMessageListener1"/>
<!--目前还没找到DefaultMessageListenerContainer配置多个监听的代码-->
<!--<bean id="topicMessageListener2" class="com.zy.mq.service.TopicMessageListener2"/>-->
<!-- 消息监听器 -->
<bean id="topicListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"></property>
<property name="destination" ref="myTopicDestination"></property>
<property name="messageListener" ref="topicMessageListener1"></property>
</bean>
</beans>
2.3 applicationContext.xml加载MQ配置
<import resource="activeMq.xml"/>
2.4 服务类
2.4.1 队列生产者 QueueProducerService
@Service
public class QueueProducerService {
@Resource(name="jmsQueueTemplate")
private JmsTemplate jmsTemplate;
public void sendMessage(Destination destination, final String msg){
System.out.println(Thread.currentThread().getName()+" 向队列"+destination.toString()+"发送消息---------------------->"+msg);
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
});
}
}
2.4.2 队列消费者QueueConsumerService
@Service
public class QueueConsumerService {
@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;
}
}
2.4.3 队列监听者QueueMessageListener
public class QueueMessageListener implements MessageListener{
@Override
public void onMessage(Message message) {
TextMessage tm = (TextMessage) message;
try {
System.out.println("QueueMessageListener监听到了文本消息:\t"
+ tm.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
2.4.4 主题生产者TopicProducerService
@Service
public class TopicProducerService {
@Resource(name="jmsTopicTemplate")
private JmsTemplate jmsTemplate;
public void sendMessage(Destination destination, final String msg){
System.out.println(Thread.currentThread().getName()+" 向主题"+destination.toString()+"发送消息---------------------->"+msg);
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
});
}
}
2.4.5 主题监听者TopicMessageListener1
public class TopicMessageListener1 implements MessageListener {
@Override
public void onMessage(Message message) {
TextMessage tm = (TextMessage) message;
try {
System.out.println("TopicMessageListener1监听到了文本消息:\t"
+ tm.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
2.5 控制类
2.5.1 QueueContorller
@RequestMapping("/queueContorller")
@Controller
public class QueueContorller {
//队列消费者
@Autowired
QueueConsumerService queueConsumerService;
//队列生产者
@Autowired
QueueProducerService queueProducerService;
@Resource(name = "myQueueDestination")
private Destination destination;
@RequestMapping("/sendQueue")
@ResponseBody
public String sendQueueMessage(){
queueProducerService.sendMessage(destination, "我向MQ服务器发送一条Queue消息");
return "sendQueue success";
}
@RequestMapping("/receiveQueue")
@ResponseBody
public String receiveQueueMessage() throws Exception{
TextMessage receive = queueConsumerService.receive(destination);
return receive.getText();
}
}
2.5.2 TopicController
@RequestMapping("/topicController")
@Controller
public class TopicController {
@Autowired
TopicProducerService topicProducerService;
@Resource(name = "myTopicDestination")
private Destination destination;
@RequestMapping("/sendTopic")
@ResponseBody
public String sendTopicMessage(){
topicProducerService.sendMessage(destination, "我向MQ服务器发送一条Topic消息");
return "sendTopicMessage success";
}
}
3.使用
Number Of Pending Messages :等待消费的消息 这个是当前未出队列的数量。
Number Of Consumers :消费者 这个是消费者端的消费者数量
Messages Enqueued :进入队列的消息 进入队列的总数量,包括出队列的。
Messages Dequeued :出了队列的消息 可以理解为是消费这消费掉的数量。
3.1 生产一个队列,然后监听器消费一个队列
3.2 生产一个主题消息,监听器消费主题消息
4.集成protobuf
4.1 添加依赖
<!--添加protobuf依赖-->
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>3.5.1</version>
</dependency>
4.2 配置监听器
<bean id="topicMessageListener2" class="com.zy.mq.service.TopicMessageListener2"/>
4.3 添加proto描述文件
4.4 编写相关类
@Service
public class TopicProducerService {
@Resource(name="jmsTopicTemplate")
private JmsTemplate jmsTemplate;
public void sendObjectMessage(Destination destination){
System.out.println(Thread.currentThread().getName()+" 向主题"+destination.toString()+"发送protobuf消息---------------------->");
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
ObjectMessage objectMessage = session.createObjectMessage();
PersonModel.Person.Builder builder = PersonModel.Person.newBuilder();
builder.setPid(2);
builder.setPname("德玛");
builder.setScore(88.9);
PersonModel.Person person = builder.build();
objectMessage.setObject(person);
return objectMessage;
}
});
}
}
public class TopicMessageListener2 implements MessageListener {
@Override
public void onMessage(Message message) {
ObjectMessage om = (ObjectMessage) message;
try {
PersonModel.Person person = (PersonModel.Person)om.getObject();
System.out.println("TopicMessageListener2监听到了文本消息:\t"
+ person.getPid()+"*"+person.getPname()+"*"+person.getScore());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
@RequestMapping("/topicController")
@Controller
public class TopicController {
@Autowired
TopicProducerService topicProducerService;
@Resource(name = "myTopicDestination")
private Destination destination;
@RequestMapping("/sendObjTopic")
@ResponseBody
public String sendObjTopicMessage(){
topicProducerService.sendObjectMessage(destination);
return "sendObjTopicMessage success";
}
}
参考
Java消息队列-Spring整合ActiveMq
ActiveMQ和spring整合,订阅主题和消息消费
ActiveMQ整合spring实现持久化消息接收
Spring与ActiveMQ整合
深入理解PooledConnectionFactory CachingConnectionFactory SingleConnectionFactory
ActiveMQ笔记之ConnectionFactory
消息中间件的基本介绍