ActiveMQ

背景

相信各位一定在网上买过东西、收过快递。在过去,快递员需要送到你家门口而你恰好有事不在家,所以快递员只能拿着快递给你打电话最后只能在你家门口等待。很大的可能就是你需要跑到快递点自己取快递。这种情况我们称之为同步通信。
现如今出现了类似蜂巢的快递寄存服务,快递员大可以将你的快递放到蜂巢等你自己回来取快递。快递员再也不用担心客户不在家的情况。这种不需要等待服务端是否处理完毕,甚至不需要知道消息是否被投递完成,我们称之为异步消息。
与之类似的的我们将快递柜称为消息队列中间件。

消息队列中间件

消息队列中间件是分布式系统中重要的组件,主要解决应用解耦,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构。目前使用较多的消息队列有ActiveMQ,RabbitMQ,ZeroMQ,Kafka,MetaMQ,RocketMQ.
我们这里主要介绍ActiveMQ

ActiveMQ

ActiveMQ 是Apache出品,最流行的,能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现,尽管JMS规范出台已经是很久的事情了,但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。–百度百科

JMS(Java Messaging Service)是Java平台上有关面向消息中间件(MOM)的技术规范,它便于消息系统中的Java应用程序进行消息交换,并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发。JMS只是定义了一套接口和规范但是并未去真正实现,ActiveMQ则是Java消息服务JMS的具体实现。

JMS消息通常有两种类型

  • 点对点消息模型(Point-Point),点对点模型中,每一条消息都有一个发送者和接受者,上例中的送快递就是点对点模型
  • 发布订阅模型(Publish-to-Point),发布订阅模型中,消息会发布一个主题,多个接收者可以订阅这个主题。而在现实中,比如微博关注,观看新闻都可以看做这类模型

如何使用ActiveMQ

官网根据系统选择版本下载ActiveMQ软件。笔者这里下载的是Windows最新版。
进行 apache-activemq-5.14.0binwin64 目录 启动 activemq.bat 文件,浏览器访问http://localhost:8161/admin/,账号密码都是admin。

Active使用的是标准生产者和消费者模型

  • Queue队列,点对点消息模型
  • Topic话题,发布订阅模型

Java操作ActiveMQ

1.导入activemq-all-5.14.0.jar,maven坐标

1
2
3
4
5
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.14.0</version>
</dependency>

2.编写MQ消息生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public class {
public static void main(String[] args) throws Exception {
//使用默认用户名密码
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
//获取一个连接
Connection connection = connectionFactory.createConnection();
// 建立会话
Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
// 创造会话或者话题对象
Queue queue = session.createQueue("HelloWorld");
//创建生产者
MessageProducer producer = session.createProducer(queue);
//发送消息
for(int i=0;i<10;i++){
producer.send(session.createTextMessage("Hello,activeMQ"));
}
session.commit();
}
}


3.编写MQ消费者代码
a)使用MessageConsumer完成消费

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
public class ConsumerMq {
public static void main(String[] args) throws Exception {
//连接工厂
//使用默认用户名密码
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
//获取一个连接
Connection connection = connectionFactory.createConnection();
// 开启连接
connection.start();
// 建立会话
// 第一个参数,是否开启事务,如果设置为true,必须session.commit();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创造会话或者话题对象
Queue queue = session.createQueue("HelloWorld");
// 创建消费者
MessageConsumer messageConsumer = session.createConsumer(queue);
while(true){
TextMessage textMessage = (TextMessage) messageConsumer.receive(10000);
if(textMessage!=null){
System.out.println(textMessage.getText());
}else{
break;
}
}
}
}

消息已经被消费
b)使用监听器完成消费

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
public class ListenerConsumerMq {
public static void main(String[] args) throws Exception {
// 连接工厂
// 使用默认用户名密码
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
// 获取一个连接
Connection connection = connectionFactory.createConnection();
// 开启连接
connection.start();
// 建立会话
// 第一个参数,是否开启事务,如果设置为true,必须session.commit();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 创造会话或者话题对象
Queue queue = session.createQueue("HelloWorld");
// 创建消费者
MessageConsumer messageConsumer = session.createConsumer(queue);
messageConsumer.setMessageListener(new MessageListener() {
//每次接受消息,自动调用onMessage
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println(textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
while (true) {
}
}
}

Spring整合ActiveMQ

1.配置Spring测试环境,maven坐标

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<version>4.1.7.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>4.1.7.RELEASE</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.14.0</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>4.1.7.RELEASE</version>
</dependency>

2.配置applicationContext.xml文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
<!-- 扫描包注解 -->
<context:component-scan base-package="com.fxtahe" />
<!-- 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" />
</bean>
<!-- 定义JmsTemplate的Topic类型 -->
<bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->
<constructor-arg ref="connectionFactory" />
<!-- pub/sub模型(发布/订阅) -->
<property name="pubSubDomain" value="true" />
</bean>
<!--Spring JmsTemplate 的消息生产者 end-->

针对冗长和重复的JMS代码,Spring给出JmsTemplate的解决方案,类似于Spring Data的JdbcTemplate。JmsTemplate可以创建连接、获得会话以及发送和接收消息,让我们去专注构造消息的发送和处理消息。通过设置pubSubDomain获取Queue队列或者Topic话题模板.

3.编写生产者代码
a) Queue 队列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@Component("queueSender")
public class QueueSender {
@Autowired
@Qualifier("jmsQueueTemplate")
private JmsTemplate jmsTemplate;// 通过@Qualifier修饰符来注入对应的bean
/**
* 发送一条消息到指定的队列(目标)
*
* @param queueName
* 队列名称
* @param message
* 消息内容
*/
public void send(String queueName, final String message) {
jmsTemplate.send(queueName, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});
}
}

b) Topic 话题

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@Component("topicSender")
public class TopicSender {
@Autowired
@Qualifier("jmsTopicTemplate")
private JmsTemplate jmsTemplate;
/**
* 发送一条消息到指定的队列(目标)
*
* @param queueName
* 队列名称
* @param message
* 消息内容
*/
public void send(String topicName, final String message) {
jmsTemplate.send(topicName, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});
}
}

4.编写消费者代码
a) Queue消费者

1
2
3
4
5
6
7
8
9
10
11
@Component
public class QueueReceiver1 implements MessageListener {
public void onMessage(Message message) {
try {
System.out.println("QueueReceiver1接收到消息:" + ((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}

1
2
3
4
5
6
7
8
9
10
11
12
@Component
public class QueueReceiver2 implements MessageListener {
public void onMessage(Message message) {
try {
System.out.println("QueueReceiver2接收到消息:" + ((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}

b) Topic消费者

1
2
3
4
5
6
7
8
9
10
11
12
@Component
public class TopicReceiver1 implements MessageListener {
public void onMessage(Message message) {
try {
System.out.println("TopicReceiver1接收到消息:" + ((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}

1
2
3
4
5
6
7
8
9
10
11
12
@Component
public class TopicReceiver2 implements MessageListener {
public void onMessage(Message message) {
try {
System.out.println("TopicReceiver2接收到消息:" + ((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}

5.配置消费者xml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
<!-- 消息消费者 start-->
<!-- 定义Queue监听器 -->
<jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
<jms:listener destination="test.queue" ref="queueReceiver1"/>
<jms:listener destination="test.queue" ref="queueReceiver2"/>
</jms:listener-container>
<jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
<jms:listener destination="test.map" ref="queueReceiver3"/>
</jms:listener-container>
<!-- 定义Topic监听器 -->
<jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
<jms:listener destination="test.topic" ref="topicReceiver1"/>
<jms:listener destination="test.topic" ref="topicReceiver2"/>
</jms:listener-container>
<!-- 消息消费者 end -->

6.测试
测试代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:applicationContext.xml")
public class ActiveMQTest {
@Autowired
private QueueSender queueSender;
@Autowired
private TopicSender topicSender;
@Test
public void testMQ() {
queueSender.send("test.queue", "Hello,Queue");
topicSender.send("test.topic", "Hello,Topic");
while (true) {
// 持续执行 ,让监听器 有时间接收信息
}
}
}

运行结果:

注意:Topic 有有效期,过了有效期后,订阅消费者无法再获取到 topic 信息


以上就是所有案例,虽然了解了用法但还需深入学习,或许下一次会有更好的见解。