ActiveMQ--消息的可靠性

落日映苍穹つ 2021-08-13 16:14 430阅读 0赞

### 持久性（PERSISTENT） ###

*  **messageProducer.setDeliverMode(DeliverMode.NON\_PERSISTENT);** 非持久化：当服务器宕机，消息不存在
 *  **messgageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);** 持久化：当服务器宕机，消息依然存在。

队列（queue）的持久化：

队列的默认为持久化模式，此模式保证这些消息只被传送一次和成功使用一次。对于这些消息，**可靠性是优先考虑的因素。**

可靠性的另一个重要方面是确保持久性消息传送到目标后，消息服务在向消费者传送它们之前不会丢失这些消息。

主题（topic）的持久化：  
消费方代码：

package pres.zhang.persistent;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import java.io.IOException;
    
    public class Topic_Consumer { 
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String TOPIC_NAME = "topic_acton";
    
        public static void main(String[] args) throws JMSException, IOException { 
            //1.创建连接工厂,按照给定的url地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工厂，获得连接connection，并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            //connection.start();
    
            //设置订阅用户
            connection.setClientID("test");
    
            //3.创建回话session
            //两个参数 1.事务 2.签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（具体是队列还是主题topic）
            Topic topic = session.createTopic(TOPIC_NAME);
    
            //创建持久化的订阅 参数：1.主题 2.备注
            TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic, "remard");
    
            //发布订阅
            connection.start();
            Message message = topicSubscriber.receive();
    
            while (null != message){ 
                TextMessage textMessage = (TextMessage) message;
                System.out.println("收到的持久化topic：" + ((TextMessage) message).getText());
                //继续监听
                message = topicSubscriber.receive(1000L);
            }
    
            session.close();
            connection.close();
        }
    }

生产方代码：

package pres.zhang.persistent;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    
    public class Topic_Produce { 
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String TOPIC_NAME = "topic_acton";
    
        public static void main(String[] args) throws JMSException { 
            //1.创建连接工厂,按照给定的url地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            //2.通过连接工厂，获得连接connection，并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            //connection.start();
    
            //3.创建回话session
            //两个参数 1.事务 2.签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4.创建目的地（具体是队列还是主题topic）
            Topic topic = session.createTopic(TOPIC_NAME);
            //创建消息的生产者
            MessageProducer messageProducer = session.createProducer(topic);
            //通过使用messageProducer生产3条消息发送到MQ的队列里面
    
            //设置持久化主题
            messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
            //因为设置持久化主题，所以connect.start()在这里调用
            connection.start();
            for (int i = 0; i < 3; i++) { 
                //7.创建消息
                TextMessage textMessage = session.createTextMessage("TOPIC_NAME---" + i);//可以理解为字符串
                //8.通过messageProducer发送给MQ
                messageProducer.send(textMessage);
            }
            //9.关闭资源
            messageProducer.close();
            session.close();
            connection.close();;
            System.out.println("消息发布到TOPIC完成！");
        }
    }

测试：先运行消费方，进行主题订阅：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

显示有一个活跃的订阅者，因为并没有消息给消费者消费，所以程序一直在监听。我们关闭程序，变化如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]  
因为我们终止的程序，所以显示到了offline栏，但是我们已经成功订阅。

现在运行消息的生产者：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]

消息发送成功。我们再次运行消费者：  
控制台打印：

收到的持久化topic：TOPIC_NAME---0
    收到的持久化topic：TOPIC_NAME---1
    收到的持久化topic：TOPIC_NAME---2

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]

成功接收到消息，因为我们已经订阅了该主题。

总结：

1.  一定要先运行一次消费者，等于向MQ注册，类似我们订阅了这个主题。
2.  然后再运行生产者发送消息
3.  此时，无论消费者是否在线，都会接收到，不在线的话，下次连接的时候，会把没有收过的消息都接收下来。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/6ffc7428103645ff8bc27bf30ce7a367.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/c328b1e3a3044bc58b3b4ee3ba5fe86a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/8882a51ab097408fac1e447c54e9dab9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/dbbd9932ccb04187bdd62c720b86d0b3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/946ae5f263c44b01bb068f6b22eb397f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]: /images/20210813/0d473a71404146bf9783f6051bd42841.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]: /images/20210813/ecd2a65917c24d0cbf7d130514e63f80.png