ActiveMQ 持久订阅者,执行结果与初衷相违背,验证离线订阅者无效,问题解决

导读

最新在接触 ActiveMQ ,里面有个 持久订阅者 模块, 功能是怎么样也演示不出来效果 配置参数比较简单(配置没几个参数),消费者第一次运行时,需要指定ClientID( 此时Broker已经记录离线订阅者信息 ),在启动提供者,此时消息队列存在一条记录,然后在启动消费者,但是怎么样也获取不到消息,阿西吧~~~什么鬼,百度上一大堆,都是这样步骤,消费者端,指定以下ClientID就好了,可,想要的效果死活不出来。。。。。。

采坑之路

废话不多说,先上代码,后面再分析

消费者端代码

    public void testTopicConsumer2() throws Exception {
        //第一步:创建ConnectionFactory
        String brokerURL = "tcp://192.168.31.215:61616";
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURL);
        //第二步:通过工厂,创建Connection
        Connection connection = connectionFactory.createConnection();
        //设置持久订阅的客户端ID
        String clientId = "10086";
        connection.setClientID(clientId);
        //第三步:打开链接
        connection.start();
        //第四步:通过Connection创建session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //第五步:通过session创建Consumer
        Topic topic = session.createTopic("cyb-topic");

        //创建持久订阅的消费者客户端
        //第一个参数是指定Topic
        //第二个参数是自定义的ClientId
        MessageConsumer consumer = session.createDurableSubscriber(topic, clientId);
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                //第七步:处理信息
                if (message instanceof TextMessage){
                    TextMessage tm=(TextMessage)message;
                    try{
                        System.out.println(tm.getText());
                    }
                    catch (Exception e){
                        e.printStackTrace();
                    }
                }
            }
        });
        //session.commit();
        //第八步:关闭资源
        consumer.close();
        session.close();
        connection.close();

    }

ActiveMQ 持久订阅者,执行结果与初衷相违背,验证离线订阅者无效,问题解决

只需要制定ClientID和创建持久客户端即可

提供者端代码

   public void testTopicProducer() throws Exception {
        Connection connection = null;
        MessageProducer producer = null;
        Session session = null;
        try {
            //第一步:创建ConnectionFactory,用于连接broker
            String brokerURL = "tcp://192.168.31.215:61616";
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURL);
            //设置
            //((ActiveMQConnectionFactory) connectionFactory).setProducerWindowSize(1000);
            //第二步:通过工厂,创建Connection
            connection = connectionFactory.createConnection();
            //第三步:连接启动
            connection.start();
            //第四步:通过连接获取session会话
            //第一个参数:是否启用ActiveMQ事务,如果为true,第二个参数无用
            //第二个参数:应答模式,AUTO_ACKNOWLEDGE为自动应答
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //第五步:通过session创建destination,两种目的地:Queue、Topic
            //参数:消息队列的名称,在后台管理系统中可以看到
            Topic topic = session.createTopic("cyb-topic");
            //第六步:通过session创建MessageProducer
            producer = session.createProducer(topic);
            //producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            //第七步:创建Message
            //方式一
            //TextMessage message=new ActiveMQTextMessage();
            //message.setText("queue test");
            //方式二
            TextMessage message1 = session.createTextMessage("topic->博客园地址:https://www.cnblogs.com/chenyanbin/");
            //第八步:通过producer发送消息
            producer.send(message1);
            //session.commit();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //第九步:关闭资源
            producer.close();
            session.close();
            connection.close();
        }
    }

验证离线订阅者功能

失败的验证

ActiveMQ 持久订阅者,执行结果与初衷相违背,验证离线订阅者无效,问题解决

正确的验证方式

首先明确一点,上面的代码是没有一点问题的。为了节省时间,验证步骤和上面的差不多,不启动前两步了,直接启动第三步,也就是:

  1. 先启动消费者(记录持久订阅者ClientID);
  2. 在启动提供者;
  3. 启动消费者( 在下面加个死循环 )

ActiveMQ 持久订阅者,执行结果与初衷相违背,验证离线订阅者无效,问题解决

问题剖析

ActiveMQ 持久订阅者,执行结果与初衷相违背,验证离线订阅者无效,问题解决

第一次运行消费者时,此时Broker已经记录订阅者ClientID,然后程序一闪而过,进入到蓝色框中的,离线订阅者中,然后在执行提供者,此时,Topic中,已经入队一次,再次运行消费者时,运行是异步获取的,运行一闪而过( 鄙人猜测,可能是ActiveMQ机制问题,内部逻辑大概是,先遍历非持久订阅者,然后在查看持久订阅者,问题出在,程序执行太快,还没到查看持久订阅者时,程序就执行完了,所以第二次执行消费者时,加了个死循环,不停监听队列消息,具体ActiveMQ底层代码没看过,有兴趣的可以研究下,底层代码找到相应位置后,记得告诉我哦~~~ )

ActiveMQ 持久订阅者,执行结果与初衷相违背,验证离线订阅者无效,问题解决

这个小问题,捣鼓一下午,百度上也说,就这2步骤配置即可,运行结果与初衷相违背,大半夜的都打算洗洗睡了,头脑风暴想出来这个方法,在下面写个死循环,不停监听队列消息,这才有了这篇博客,好啦…好啦,时间不早了,马上都快凌晨1点钟了,明个还得上班,洗洗睡了zZZZZZZZZZ

原文 

http://www.cnblogs.com/chenyanbin/p/12866452.html

本站部分文章源于互联网,本着传播知识、有益学习和研究的目的进行的转载,为网友免费提供。如有著作权人或出版方提出异议,本站将立即删除。如果您对文章转载有任何疑问请告之我们,以便我们及时纠正。

PS:推荐一个微信公众号: askHarries 或者qq群:474807195,里面会分享一些资深架构师录制的视频录像:有Spring,MyBatis,Netty源码分析,高并发、高性能、分布式、微服务架构的原理,JVM性能优化这些成为架构师必备的知识体系。还能领取免费的学习资源,目前受益良多

转载请注明原文出处:Harries Blog™ » ActiveMQ 持久订阅者,执行结果与初衷相违背,验证离线订阅者无效,问题解决

赞 (0)
分享到:更多 ()

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址