AMQ的持久化问题

前言

​ 前面一篇AMQ专题中,我们发现对于Topic这种类型的消息,即使将deliveryMode设置为持久化,只要生产者在消费者之前启动。消息生产者发布的消息还是会丢失。这是符合JMS规范的。

当然,作为一个如此活跃的开源消息中间件,在实现JMS基本规范之后,必然会通过扩展的方式来实现Topic的持久化订阅。

而所谓的deliveryMode持久化和订阅持久化还是两个不同的概念。本篇博客我们就通过实例来一探究竟。

DeliveryMode持久化

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

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

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

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

创建producer,并将producer的deliveryMode设置成持久化,运行producer

在消息被consumer消费之前,重启activeMQ

运行consumer,发现接收到了activeMQ重启之前Producer发送的消息

修改producer,将producer的deliveryMode设置成非持久化,运行producer

在消息被consumer消费之前,重启activeMQ

运行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) {

                }

            }

        }

    }

}

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

运行producer,向activeMQ发送主题消息

运行consumer,发现未收到任何消息

运行producer,此时运行中的consumer接收到了topic消息

停止运行consumer,重新运行producer

重新运行consumer,此时consumer接收到了刚刚producer产生的消息

创建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的持久化和订阅持久化是两个不同的概念,二者互不干扰,组合实现业务需求

需要弄清参数的实际意义第一步自己动手写实例,看运行结果是否与自己预期一致。第二步则是情况允许的时候,多看源码,掌握好的代码和设计

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 219,539评论 6 508
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 93,594评论 3 396
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 165,871评论 0 356
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 58,963评论 1 295
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 67,984评论 6 393
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 51,763评论 1 307
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 40,468评论 3 420
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 39,357评论 0 276
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 45,850评论 1 317
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 38,002评论 3 338
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 40,144评论 1 351
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 35,823评论 5 346
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 41,483评论 3 331
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 32,026评论 0 22
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 33,150评论 1 272
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 48,415评论 3 373
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 45,092评论 2 355

推荐阅读更多精彩内容

  • ActiveMQ 即时通讯服务 浅析http://www.cnblogs.com/hoojo/p/active_m...
    bboymonk阅读 1,490评论 0 11
  • Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具(例如配置管理,服务发现,断路器,智...
    卡卡罗2017阅读 134,672评论 18 139
  • 个人专题目录[https://www.jianshu.com/p/140e2a59db2c] 一、JMS简介 全称...
    Java及SpringBoot阅读 2,087评论 0 10
  • Kafka简介 Kafka是一种分布式的,基于发布/订阅的消息系统。主要设计目标如下: 以时间复杂度为O(1)的方...
    Alukar阅读 3,083评论 0 43
  • 请没收孩子的手机 那里有太多不良的信息 请没收孩子的手机 那里有低级的视频与暴力的游戏 请没收孩子的手机 他们鉴别...
    刘军_00ee阅读 124评论 0 0