一、简介
1.什么是消息中间体
消息中间件利用高效可靠的消息传递机制进行平台无关的数据交流,并基于数据通信来进行分布式系统的集成。通过提供消息传递和消息排队模型,它可以在分布式环境下扩展进程间的通信。对于消息中间件,常见的角色有Producer(生产者)、Consumer(消费者)
2.什么是JMS
JMS(Java Messaging Service)是Java平台上有关面向消息中间件的技术规范,它便于消息系统中的Java应用程序进行消息交换,并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发。
JMS 定义了五种不同的消息正文格式,以及调用的消息类型,允许你发送并接收以一些不同形式的数据,提供现有消息格式的一些级别的兼容性。
TextMessage--一个字符串对象 MapMessage--一套名称-值对 ObjectMessage--一个序列化的 Java 对象 BytesMessage--一个字节的数据流 StreamMessage -- Java 原始值的数据流
3.JMS消息传递类型
对于消息的传递有两种类型:
一种是点对点的,即一个生产者和一个消费者对应
一种是发布/订阅模式,即一个生产者产生消息并进行发送后,可以由多个消费者进行接收
4.ActiveMQ下载与安装
(1)下载
官方网站下载:http://activemq.apache.org/download.html
(2)安装启动(Linux)
1.将apache-activemq-版本号-bin.tar.gz上传至服务器
2.解压此文件
tar zxvf apache-activemq-5.12.0-bin.tar.gz
3.为apache-activemq目录赋权
chmod 777 apache-activemq-5.12.0
4.启动
bin目录下运行: ./activemq start
二、Spring整合JMS
1.点对点模式
(1)消息生产者
①创建工程springjms_product
②导入pom依赖
<properties>
<spring.version>4.2.4.RELEASE</spring.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>${spring.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${spring.version}</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.9</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
<version>5.13.4</version>
</dependency>
</dependencies>
③创建spring配置文件
applicationContext-jms-producer.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jms="http://www.springframework.org/schema/jms"
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">
<context:component-scan base-package="com.yfy"></context:component-scan>
<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://192.168.25.132:61616"/>
</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="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="queue_text"/>
</bean>
</beans>
④创建生产者类
QueueProduct.java
@Component
public class QueueProduct {
@Autowired
private JmsTemplate jmsTemplate;
@Autowired
private Destination queueTextDestination;
public void sentTextMessag(final String message) {
jmsTemplate.send(queueTextDestination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});
}
}
⑤创建测试类
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("classpath:applicationContext-jms-producer.xml")
public class Test {
@Autowired
private QueueProduct queueProduct;
@org.junit.Test
public void testSendQueue() {
queueProduct.sentTextMessag("hello queue");
}
}
(2)消息消费者
①创建工程springjms_consumer
②导入pom依赖,和消息生产者一样
③创建spring配置文件
applicationContext-jms-consumer.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jms="http://www.springframework.org/schema/jms"
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">
<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供-->
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://192.168.25.132:61616"/>
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="targetConnectionFactory"/>
</bean>
<!--这个是队列目的地,点对点的 文本信息-->
<bean id="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="queue_text"/>
</bean>
<!--这个是订阅模式 文本信息-->
<bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="topic_text"/>
</bean>
<!-- 我的监听类 -->
<bean id="myMessageListener" class="com.yfy.MyMessageListener"></bean>
<!-- 消息监听容器:队列 -->
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="queueTextDestination" />
<property name="messageListener" ref="myMessageListener" />
</bean>
</beans>
④创建监听类
MyMessageListener.java
public class MyMessageListener implements MessageListener {
@Override
public void onMessage(Message message) {
TextMessage testMessage=(TextMessage)message;
try {
String text = testMessage.getText();
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
}
⑤创建测试类
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("classpath:applicationContext-jms-consumer.xml")
public class Test {
@org.junit.Test
public void testReceive(){
try {
System.in.read();
} catch (IOException e) {
e.printStackTrace();
}
}
}
2.发布/订阅模式
(1)消息生产者
在product配置文件中加入
<!--这个是订阅模式 文本信息-->
<bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="topic_text"/>
</bean>
TopicProduct.java
@Component
public class TopicProduct {
@Autowired
private JmsTemplate jmsTemplate;
@Autowired
private Destination topicTextDestination;
public void sendMessage(String message) {
jmsTemplate.send(topicTextDestination, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});
}
}
创建测试类
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("classpath:applicationContext-jms-producer.xml")
public class Test {
@Autowired
private TopicProduct topicProduct;
@org.junit.Test
public void TestSendTopic() {
topicProduct.sendMessage("hello topic");
}
}
(2)消息消费者
在consumer配置文件中加入
<!-- 我的监听类 -->
<bean id="myMessage2Listener" class="com.yfy.MyMessage2Listener"></bean>
<!-- 消息监听容器:订阅模式 -->
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="topicTextDestination" />
<property name="messageListener" ref="myMessageListener" />
</bean>
创建监听类
MyMessage2Listener.java
public class MyMessage2Listener implements MessageListener {
@Override
public void onMessage(Message message) {
TextMessage textMessage=(TextMessage)message;
try {
String text = textMessage.getText();
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
}
测试:同时运行三个消费者工程,在运行生产者工程,查看消费者工程的控制台输出