ActiveMQ专题2: 持久化

AMQ的持久化问题

前言

​ 前面一篇AMQ专题中,我们发现对于Topic这种类型的消息,即使将deliveryMode设置为持久化,只要生产者在消费者之前启动。消息生产者发布的消息还是会丢失。这是符合JMS规范的。 当然,作为一个如此活跃的开源消息中间件,在实现JMS基本规范之后,必然会通过扩展的方式来实现Topic的持久化订阅。 而所谓的deliveryMode持久化和订阅持久化还是两个不同的概念。本篇博客我们就通过实例来一探究竟。

DeliveryMode持久化

​ 在前面一篇中,我们通过producer.setDeliveryMode(DeliveryMode.PERSISTENT);将消息传递特性置为持久化,当时尝试过当消息类型是topic的时候,不管该值设置为啥,只要先启动Producer,那么对于后启动的Consumer都无法获取原来发布的主题。

​ 那么这个DeliveryMode究竟是用来干啥的呢?

  • DeliveryMode中的是否持久化,指的是当重启activeMQ之后,原来队列或者主题中未被消费的消息是否仍然保留

​ 我这里自己通过代码进行了如下测试,测试步骤和结果如下:

  1. 创建producer,并将producer的deliveryMode设置成持久化,运行producer
  2. 在消息被consumer消费之前,重启activeMQ
  3. 运行consumer,发现接收到了activeMQ重启之前Producer发送的消息
  4. 修改producer,将producer的deliveryMode设置成非持久化,运行producer
  5. 在消息被consumer消费之前,重启activeMQ
  6. 运行consumer,没有接收到任何消息,原producer产生的消息丢失

​ 持久化和非持久化最终队列控制台分别如下:

     至此,不难发现,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) {
                }
            }
        }
    }
}

​ 最终我的验证步骤和结果如下:

  1. 运行producer,向activeMQ发送主题消息
  2. 运行consumer,发现未收到任何消息
  3. 运行producer,此时运行中的consumer接收到了topic消息
  4. 停止运行consumer,重新运行producer
  5. 重新运行consumer,此时consumer接收到了刚刚producer产生的消息
  6. 创建consumer的session的时候,同时创建两个同clientId的session时会报同一通道已被占用的错误

​ 分析以上步骤,我最终对这种伪持久化订阅的总结如下:

  • 要实现伪持久化订阅,必须先向activeMQ发布持久化订阅消息,通过clientId来标识不同的订阅渠道。 如果在发布持久化订阅消息之前producer就向mq发送了topic消息,那么consumer还是没法接收
  • activeMQ确定是否是同一持久化订阅者的依据条件有两个:connection.setClientID("3")中的clientId 以及session.createDurableSubscriber(session.createTopic("KiDe-topic-Demo"), "12")中的name

总结

  • deliveryMode的持久化和订阅持久化是两个不同的概念,二者互不干扰,组合实现业务需求
  • 需要弄清参数的实际意义第一步自己动手写实例,看运行结果是否与自己预期一致。第二步则是情况允许的时候,多看源码,掌握好的代码和设计

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏北京马哥教育

如何快速截取某段时间内的日志

在排除故障时,需要分析发生故障的原因,避免再次出现同样的问题,需要对日志进行分析,截取故障前后时间段的日志。通常用正则表达式来获取某段时间的内的日志,比较繁琐。...

30260
来自专栏测试开发架构之路

centos7安装python3.6后导致防火墙功能无法正常工作的解决办法

问题:因为默认python版本被设置成了python3.6,而进行防火墙的指令操作频频报错。

9810
来自专栏散尽浮华

线上图片批量更换脚本记录

需求说明: 线上之前使用的图片没有经过处理,比较大。应产品同事要求,将图片拿出来处理后再替换之前的图片,图片名不变。 /tmp/image目录下存放的修改后的图...

22060
来自专栏Danny的专栏

【EJB学习笔记】——JMS和消息驱动Bean

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/huyuyang6688/article/...

11220
来自专栏Java开发者杂谈

ActiveMQ专题1: 入门实例

​ 从上面的代码可以看出,生产者和消费者的处理流程大致相同。存在很多重复代码,不难发现可以抽取出公共的代码来使得代码更加简洁。

12720
来自专栏公众号_薛勤的博客

Java消息中间件入门笔记 - ActiveMQ篇

1.下载安装包 wget http://120.221.32.78:6510/mirrors.shu.edu.cn/apache//activemq/5.1...

48720
来自专栏编程

ActiveMQ消息队列的使用及应用

ActiveMQ消息队列的使用及应用 这里就不说怎么安装了,直接解压出来就行了。 谢绝转载,作者保留所有权力 目录: 复制代码 一:JMQ的两种消息模式 1.1...

35370
来自专栏Petrichor的专栏

macOS: 没有移动硬盘的 写权限

Note: 这里以希捷(seagate)硬盘为例。其他牌子移动硬盘写权限的问题可以举一反三。

44040
来自专栏kangvcar

关于Linux上SAMBA服务的权限问题(普通挂载)

21330
来自专栏积累沉淀

干货--JMS(java消息服务)整合Spring项目案例

Sprng-jms消息服务小项目 所需的包: spring的基础包 spring-jms-xx包 spring-messag...

564100

扫码关注云+社区

领取腾讯云代金券