JMS介绍（二）
2017-09-16 本文已影响0人
guqj
只要你不计较得失，人生还有什么不能想法子克服的。 —— 海明威
JMS的消费模型分为Point-to-Point（点对点）和Publish/Subscribe（发布/订阅）两种类型。
1. PTP模型
1.1 简介
PTP模型是基于队列的，生产者发消息到队列，消费者从队列接收消息，队列的存在使得消息的异步传输成为可能。如下图所示：
PTP模型图
1.2 特点
- 每个消息只有一个消费者（Consumer）(即一旦被消费，消息就不再在消息队列中)。
- 当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列。
- 接收者在成功接收消息之后需向队列应答acknowledged)成功。
- 如果在Session 关闭时，有一些消息已经被收到，但还没有被应答(acknowledged)，那么，当消费者下次连接到相同的队列时，这些消息还会被再次接收。
- 如果用户在receive方法中设定了消息选择条件，那么不符合条件的消息会留在队列中，不会被接收到。
- 队列可以长久地保存消息直到消费者收到消息。
1.3 示例
// 消息发送者
public void produce(){
// 实例化连接工厂
ConnectionFactory connectionFactory = new
ActiveMQConnectionFactory("admin", "admin", "failover://tcp://localhost:61616");
// 通过连接工厂获取连接
Connection connection = connectionFactory.createConnection();
// 启动连接
connection.start();
// 创建session
Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
// 创建队列
Destination destination = session.createQueue(queue);
// 创建生产者
MessageProducer messageProducer =
session.createProducer(destination);
// 创建TextMessage消息
TextMessage textMessage = session.createTextMessage("Jms-Queue");
// 向队列发送消息
messageProducer.send(textMessage);
}
// 消息消费者
public void consume(){
// 实例化连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "failover://tcp://localhost:61616");
// 通过连接工厂获取连接
Connection connection = connectionFactory.createConnection();
// 启动连接
connection.start();
// 创建session
Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);
// 创建队列
Destination destination = session.createQueue("MyQueue@1");
// 创建消费者
MessageConsumer messageConsumer = session.createConsumer(destination);
// 从队列中获取消息
Message message = messageConsumer.receive(10000);
System.out.println("接收到的消息："+ ((TextMessage) message).getText());
}
2. Pub/Sub模型
2.1 简介
Pub/Sub模型定义了如何向一个内容节点发布和订阅消息，这些节点被称作topic主题可以被认为是消息的传输中介，发布者(publisher)发布消息到主题，订阅者(subscribe) 从主题订阅消息。如下图所示：
Pub/Sub模型图
2.2 特点
- 每个topic主题可以有多个消费者。
- 针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息。
- 消息订阅分为非持久订阅和持久订阅
非持久订阅：只有当客户端和JMS Provider保持连接状态才能收到Topic的消息。当客户端处于离线状态期间，则不会收到Topic的消息。
持久订阅：客户端向JMS Provider注册一个识别身份的ID，JMS Provider会为这个ID保存所有发送到Topic的消息，当客户端再次连接到JMS Provider时，会根据ID获取到所有处于离线期间发送到Topic的消息。 - 非持久订阅状态下，不能恢复或重新派送一个未应答(acknowledged)的消息。只有持久订阅才能恢复或重新派送一个未应答(acknowledged)的消息。
- 当所有的消息必须被接收，则用持久订阅。当丢失消息能够被容忍，则用非持久订阅。
2.3 示例
// 发送消息到Topic
public void publish(){
// 实例化连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "failover://tcp://localhost:61616");
// 通过连接工厂获取连接
Connection connection = connectionFactory.createConnection();
// 启动连接
connection.start();
// 创建session
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
// 创建队列
Destination destination = session.createTopic("MyTopic@1");
// 创建Topic
MessageProducer messageProducer =
session.createProducer(destination);
// 设置为持久订阅
messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
// 创建TextMessage消息
TextMessage textMessage = session.createTextMessage("发送消息：来自于xx的消息");
// 向Topic发送消息
messageProducer.send(textMessage);
}
// 订阅Topic
public void subscribe(){
// 实例化连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "admin", "failover://tcp://localhost:61616");
// 通过连接工厂获取连接
Connection connection = connectionFactory.createConnection();
// 启动连接
connection.start();
// 创建session
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
// 创建Topic
Destination destination = session.createTopic("MyTopic@1");
// 创建消费者
MessageConsumer messageConsumer = session.createConsumer(destination);
// 从队列中获取消息
Message message = messageConsumer.receive(10000);
System.out.println("接收到的消息："+ ((TextMessage) message).getText());
}
3. JMS可靠性
3.1 消息接收确认
JMS消息只有在被确认后，才认为成功的被消费。消息的成功消费通常包含三个阶段：客户接收消息、客户处理消息和消息被确认。
- 事务性会话
在事务性会话中，当一个事务被提交后，自动确认。
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
.....
session .commit();
- 非事务性会话
在非事务性会话中，消息何时被确认取决于创建会话时的应答模式。该参数有以下三个模式：
（1） Session.AUTO_ACKNOWLEDGE
当客户成功的从receive()方法返回的时候，或者从MessageListener.onMessage()方法成功返回的时候，会话自动确认客户收到的消息。
Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
...
MapMessage message = (MapMessage) consumer.receive();
（2） Session.CLIENT_ACKNOWLEDGE
客户通过调用消息的acknowledge方法确认消息。
需要注意的是，在这种模式中，确认是在会话层上进行，确认一个被消费的消息将自动确认所有已被会话消费的消息。
如：若一个消费者消费了10个消息，然后确认第10个消息，那么10个消息都被确认。
（3） Session.DUPS_OK_ACKNOWLEDGE
该模式只是会话迟钝的确认消息的提交。
3.2 消息持久性
JMS支持以下两种消息提交模式：
PERSISTENT：指示JMS provider持久保存消息，以保证消息不会因为JMSprovider的失败而丢失。
NON_PERSISTENT：不要求JMS provider持久保存消息。
3.3 消息过期
可以设置消息在一定时间后过期，默认是永不过期。
3.4 消息的临时目的地
可通过createTemporaryQueue()方法和createTemporaryTopic()方法来创建临时目的地。临时目的地存在的时间只限于创建它们的连接所保持的时间。
只有创建该临时目的地的连接上的消息消费者才能够从临时目的地中接收消息。
3.5 本地事务
在JMS客户端，可以使用本地事务来组合消息的接收与发送。JMS Session接口提供了commit和rollback方法。
（1）事务提交意味着生产的所有消息被发送，消费的所有消息被确认。
（2）事务回滚意味着生产的所有消息被销毁，消费的所有消息被恢复并重新提交，除非它们已经过期。
