activemq中持久订阅者和非持久订阅者区别

太过爱你忘了你带给我的痛 2021-06-24 15:58 450阅读 0赞

1、消息的传输类型：

上文已经讲过，activemq支持两只传输的模式：持久化和非持久化。可以通过MessageProducer 类的 setDeliveryMode方法设置传输模式：

MessageProducer producer = ...;
    producer.setDeliveryMode(DeliveryMode.PERSISTENT);

持久传输和非持久传输最大的区别是：采用持久传输时，传输的消息会保存到磁盘中(messages are persisted to disk/database)，即“存储转发”方式。先把消息存储到磁盘中，然后再将消息“转发”给订阅者。当Borker宕机 恢复后，消息还在。

采用非持久传输时，发送的消息不会存储到磁盘中。Borker宕机重启后，消息丢失。

比如，当生产者将消息投递给Broker后，Broker将该消息存储到磁盘中，在Broker将消息发送给Subscriber之前，Broker宕机了，如果采用持久传输，Broker重启后，从磁盘中读出消息再传递给Subscriber；如果采用非持久传输，这条消息就丢失了。

2、消息的订阅类型：

对于topic的消息，有两种订阅类型：Durable Subscribers 和 NonDurable Subscribers。

1）特点：

*  持久订阅者和非持久订阅者针对的Domain是Pub/Sub，而不是P2P
 *  当Broker发送消息给订阅者时，如果订阅者处于 inactive 状态：持久订阅者可以收到消息，而非持久订阅者则收不到消息。

当持久订阅者处于 inactive 状态时，Broker需要为持久订阅者保存消息；会造成的影响是：如果持久订阅者订阅的消息太多则会溢出。(当消息投递成功之后，Broker就可以把消息删除了)

2）举例：

例如，设想一个持久订户S在时间D1开始订阅主题T.  
某些发布者向主题发送消息M1，M2，M3并且S将接收这些消息中的每一个。  
然后S停止，出版商继续发送M4，M5。  
当S在D2重新启动时，发布者发送M6和M7。  
现在S会收到M4，M5，然后是M6和M7以及所有未来的消息。即S将接收来自M1..M7的所有消息。

3）代码：

activemq区分消费者，是通过clientID和订户名称来区分的。
    -----------------------------
    // 创建connection
    connection = connectionFactory.createConnection();
    connection.setClientID("bbb"); //持久订阅需要设置这个。
    connection.start();
    
    // 创建session
    Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
    
    // 创建destination
    Topic topic = session.createTopic("userSyncTopic"); //Topic名称
    
    //MessageConsumer consumer = session.createConsumer(topic); //普通订阅
    MessageConsumer consumer = session.createDurableSubscriber(topic,"bbb"); //持久订阅
    
    
    还有一点，消息的生产者，发送消息时用使用持久模式
    MessageProducer producer = ...;
    producer.setDeliveryMode(DeliveryMode.PERSISTENT);
    不设置，默认就是持久的

注：使用相同的“clientID”，则认为是同一个消费者。两个程序使用相同的“clientID”，则同时只能有一个连接到activemq，第二个连接的会报错。

二、**提高非持久订阅者的可靠性 以及 订阅恢复策略**：

对于持久订阅者而言，只要订阅了某个Topic，就不用担心自己“离线”(inactive)后，错过某些消息。但是对于非持久订阅者：

1）生产者发送了若干个消息到Topic后，非持久订阅者才去订阅该Topic，则它会错过(收不到)在它订阅之前发送的消息。

2）生产者向Topic发送了若干个消息，而此时因网络中断原因或者非持久订阅者宕机时，非持久订阅者刚好不在线(inactive)，就会错过（收不到）生产者发送的消息。

3）从消息的角度而言，有些消息是实时消息(如，实时股票价格)，需要快速地消费掉，对消息进行持久化就没有太大的意义，而且会因为存储消息而造成一定的开销。  
  
因此，为了提高非持久订阅者的可靠性，以及实时的消费消息，就需要：

1.  消息不进行持久化并缓存消息(Caching message for nondurable consumers)；
2.  对缓存的消息的消费策略

1、Retroactive Consumers：

Retroactive Consumer属于非持久订阅者，但它是消费 非持久化消息 的订阅者。（其他非持久订阅者 可以消费持久化消息）

1）ActiveMQ Broker可以为各种Topic缓存消息(但不支持 temporary topic 和 advisory topic)。这说明：该机制只针对Topic而言。

2）缓存的消息只会发给 retroactive consumer，并不会发送给持久订阅者。

那非持久订阅者如何成为 retroactive consumer呢？最简单的方式是在创建Topic的时候指定consumer为retroactive

Topic topic = session.createTopic("soccer.division1.leeds?consumer.retroactive=true");
    MessageConsumer consumer = session.createConsumer(topic);

2、消息“订阅恢复策略”(Subscription Recovery Policies)：

正因为 retroactive consumer 消费的是非持久化的消息（消息保存在内存中），所以就会出现 第一部分中提到的这两个问题：

1）生产者发送了若干个消息到Topic后，非持久订阅者才去订阅该Topic，则它会错过(收不到)在它订阅之前发送的消息。

就需要：**订阅恢复策略**。订阅恢复策略的目的就是让retroactive consumer能够回到过去某个时间点消费它错过了的消息。比如说：生产者发送了消息A，消息B给Broker的Topic之后，retroactive consumer才订阅该Topic，订阅恢复策略就可以让retroactive consumer能收到在它订阅之前就已经发送的消息(消息A 和消息B)  
  
  
3、订阅恢复策略主要有以下几类，简要介绍如下：（具体的配置参数可参考官网）

1）FixedSizedSubscriptionRecoveryPolicy：是ActiveMQ默认的策略。该恢复订阅策略最大的特点是：开辟多大的内存缓存发送到该Topic的消息。

2）Fixed Count Subscription Recovery Policy：按照数量来缓存消息。即，允许Topic最大缓存多少条消息。举例如下：

假设设置的Topic最大可缓存1000条消息。当前Topic已经缓存了500条消息了，retroactive consumer在 收到了一些消息之后宕机了，在宕机这段时间内，生产者又向Topic发送了100条消息。那么当retroactive consumer恢复正常后，生产者又向Topic发送了200条消息，那么：retroactive consumer 首先会收到它宕机期间错过的100条消息，然后就收到刚刚的200条消息。（总的消息条数未超过1000）  
.....  
还有其他一些恢复订阅策略就不一 一介绍了。总之，恢复订阅策略针对的是非持久化的retroactive consumer订阅者而言的。它提高了非持久化消息的可靠性。

参考：http://www.cnblogs.com/hapjin/p/5649696.html