消息队列 ActiveMQ·基础篇
一: ActiveMQ介绍
ActiveMQ 是一个 **MOM**,具体来说是一个实现了 **JMS 规范**的**系统间远程通信的消息代理**。
MOM 就是面向消息中间件(Message-oriented middleware),是用于以分布式应用或系统中的异步、松耦合、可靠、可扩展和安全通信的一类软件。**MOM 的总体思想是它作为消息发送器和消息接收器之间的消息中介**,这种中介提供了一个全新水平的松耦合
JMS 即Java消息服务(Java Message Service)应用程序接口,是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。Java消息服务是一个与具体平台无关的API,绝大多数MOM提供商都对JMS提供支持。
二:消息队列
消息队列 全称为Message Queue,消息队列(MQ)是正确而又完整的 JMS 实现,消息队列(MQ)是一种应用程序对应用程序的通信方法。
应用程序,通过写和检索出入列队的针对应用程序的数据(消息)来通信,而无需专用连接来链接它们。消息传递指的是程序之间通过
在消息中发送数据进行通信,而不是通过直接调用彼此来通信,直接调用通常是用于诸如远程过程调用的技术。
三:应用场景
- 异步处理
场景说明:新用户注册发放100积分,180元新手大礼包,**会员卡,传统的做法有两种:串行方式,并行方式。
- 串行方式
- 使用消息队列 并行方式
以上两种方式,很容易发现同步处理(串行方式)的情况下都会涉及到非主业务的其他操作,整个执行流程大概需要200ms,等待时间过长,
正常逻辑其实注册的的主流程不应该受其他事件影响,通过消息队列的方式,可以把后续的处理流程进行异步处理可以大大提高响应速度,
由原来的200ms变成现在的50ms,后面发放100积分,180元新手大礼包,**会员卡不影响主业务逻辑
2:应用解耦
场景说明: 业务系统之间合作,两套不同企业系统,会员信息对接数据共享,常常的方式都是通过三方接口来实现数据信息交换
- 传统企业:
根据业务需求修改业务逻辑,来实现数据同步,导致代码相互交叉,耦合度太高,需要花费大量的时间整合,一旦三方接口出现问题影响主业务
- 消息队列:
1.注册完成然后将消息写入队列返回成功。
2.发放权益业务不影响主业务,实现解耦
四:JMS两种消息传送模式
P2P (点对点)消息域使用 queue 作为 Destination,消息可以被同步或异步的发送和接收,每个消息只会给一个 Consumer 传送一次。
Consumer 可以使用 MessageConsumer.receive() 同步地接收消息,也可以通过使用MessageConsumer.setMessageListener() 注册一个
MessageListener 实现异步接收。多个 Consumer 可以注册到同一个 queue 上,但一个消息只能被一个 Consumer 所接收,然后由该
Consumer 来确认消息。并且在这种情况下,Provider 对所有注册的 Consumer 以轮询的方式发送消息。
Pub/Sub(发布/订阅,Publish/Subscribe)消息域使用 topic 作为 Destination,发布者向 topic 发送消息,订阅者注册接收来自 topic 的消息。
发送到 topic 的任何消息都将自动传递给所有订阅者。接收方式(同步和异步)与 P2P 域相同。
除非显式指定,否则 topic 不会为订阅者保留消息。当然,这可以通过持久化(Durable)订阅来实现消息的保存。这种情况下,当订阅者与
Provider 断开时,Provider 会为它存储消息。当持久化订阅者重新连接时,将会受到所有的断连期间未消费的消息。
queue消息生产者代码
//创建步骤
//第一步:获取工厂连接
ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
//第二步:通过工厂创建连接
Connection connection = factory.createConnection();
//开启连接
connection.start();
//创建回话session
//参数一:是否对事物支持: true 参数二值被忽略 默认值为:SESSION_TRANSACTED
//Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
//参数一:是否对事物支持: false
//Session.AUTO_ACKNOWLEDGE:为自动确认,客户端发送和接收消息不需要做额外的工作。
//Session.CLIENT_ACKNOWLEDGE:为客户端确认。客户端接收到消息后,必须调用javax.jms.Message的acknowledge方法。jms服务器才会删除消息。
//Session.DUPS_OK_ACKNOWLEDGE:允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回,
// 会话对象就会确认消息的接收;而且允许重复确认。在需要考虑资源使用时,这种模式非常有效。
//SESSION_TRANSACTED
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//创建会话对象类型
Queue queue = session.createQueue("test-queue");
//使用会话创建生产对象
MessageProducer producer = session.createProducer(queue);
//创建会话消息对象
TextMessage message = session.createTextMessage("我是test-queue测试数据");
//发送消息
producer.send(message);
producer.close();
session.close();
connection.close();
queue消息消费者代码
ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue("test-queue");
MessageConsumer consumer = session.createConsumer(queue);
//接收消息方法一:同步
Message message = consumer.receive();
System.out.println(message);
/*
接收消息方法二:异步
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println(textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});*/
//程序等待接收用户消息
System.in.read();
//关闭资源
consumer.close();
session.close();
connection.close();
topic消息生产者代码
ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic("test-topic");
MessageProducer producer = session.createProducer(topic);
//producer.setDeliveryMode(DeliveryMode.PERSISTENT);
TextMessage message = session.createTextMessage("发布订阅模式!");
producer.send(message);
producer.close();
session.close();
connection.close();
topic消息消费者代码
ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = session.createTopic("test-topic");
MessageConsumer consumer = session.createConsumer(topic);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
TextMessage text = (TextMessage)message;
System.out.println(text);
}
});
System.in.read();
consumer.close();
session.close();
connection.close();