前面一篇AMQ专题中,我们发现对于Topic这种类型的消息,即使将deliveryMode设置为持久化,只要生产者在消费者之前启动。消息生产者发布的消息还是会丢失。这是符合JMS规范的。 当然,作为一个如此活跃的开源消息中间件,在实现JMS基本规范之后,必然会通过扩展的方式来实现Topic的持久化订阅。 而所谓的deliveryMode持久化和订阅持久化还是两个不同的概念。本篇博客我们就通过实例来一探究竟。
在前面一篇中,我们通过producer.setDeliveryMode(DeliveryMode.PERSISTENT);
将消息传递特性置为持久化,当时尝试过当消息类型是topic
的时候,不管该值设置为啥,只要先启动Producer
,那么对于后启动的Consumer
都无法获取原来发布的主题。
那么这个DeliveryMode
究竟是用来干啥的呢?
我这里自己通过代码进行了如下测试,测试步骤和结果如下:
持久化和非持久化最终队列控制台分别如下:
至此,不难发现,deliveryMode的是否持久化是针对activeMQ服务器是否重启而言的。对于不支持持久化的设置,当mq重启之后,没有被消费的消息就会丢失。而支持持久化的设置,只要消息没有被消费,重启mq,仍然能被新加入的consumer消费。
JMS的规范是没有要求实现订阅持久化的。所幸的是activeMQ实现了这个特性。个人认为所谓的订阅持久化相对于消息的持久化,不过是一种伪持久化。先不做太多说明,我们直接看一个示例代码:
public class SimpleProducer {
public static void main(String[] args) {
// STEP1: 得到连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, ActiveMQConnection.DEFAULT_BROKER_URL);
Connection connection = null;
Session session = null;
MessageProducer topicProducer = null;
Destination topicDestination = null;
try {
// STEP2: 从连接工厂得到连接并且启动连接
connection = connectionFactory.createConnection();
connection.start();
// STEP3: 获取会话
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// STEP4: 创建主题
topicDestination = session.createTopic("KiDe-topic-Demo");
// STEP5: 创建消息生产者
topicProducer = session.createProducer(topicDestination);
topicProducer.setDeliveryMode(DeliveryMode.PERSISTENT); // 设置为持久化
// STEP6: 发送消息
for (int i=0; i<20; i++) {
TextMessage message = session.createTextMessage("Producer message:" + i);
topicProducer.send(message);
}
// STEP7: 如果开启了事务 ,此时需要调用session提交操作
// session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally {
if (connection != null) {
try {
connection.close();
} catch (JMSException e) {
}
}
}
}
}
public class SimpleConsumer {
public static void main(String[] args) {
// STEP1: 创建连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, ActiveMQConnection.DEFAULT_BROKER_URL);
Connection connection = null;
Session session = null;
MessageConsumer topicConsumer = null;
try {
// STEP2: 从连接工厂得到连接并且启动连接
connection = connectionFactory.createConnection();
connection.setClientID("1"); // 如果要进行持久化订阅,必须对连接设置clientID
connection.start();
// STEP3: 获取会话
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// STEP4: 创建持久化订阅者
TopicSubscriber topicSubscriber = session.createDurableSubscriber(session.createTopic("KiDe-topic-Demo"), "1");
// STEP5: 设置消息接收监听
topicSubscriber.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message paramMessage) {
TextMessage message = (TextMessage) paramMessage;
try {
System.out.println("消费者接收到主题消息:" + message.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
TimeUnit.SECONDS.sleep(200); // 睡眠200秒,使得客户端可以接收到对应消息
} catch (Exception e) {
e.printStackTrace();
} finally {
if (connection != null) {
try {
connection.close();
} catch (JMSException e) {
}
}
}
}
}
最终我的验证步骤和结果如下:
分析以上步骤,我最终对这种伪持久化订阅的总结如下:
connection.setClientID("3")
中的clientId
以及session.createDurableSubscriber(session.createTopic("KiDe-topic-Demo"), "12")
中的name