ActiveMq NON_PERSISTENT与PERSISTENT以及 durable subscription（持久订阅）的理解

灰太狼 2022-07-11 15:29 123阅读 0赞

转载至：http://quentinxxz.iteye.com/blog/2113458

实验一：

Java代码   ![收藏代码][icon_star.png]

1. public class Producer \{  
2.     public static void main(String\[\] args) \{  
3.         String user = ActiveMQConnection.DEFAULT\_USER;  
4.         String password = ActiveMQConnection.DEFAULT\_PASSWORD;  
5.         String url = ActiveMQConnection.DEFAULT\_BROKER\_URL;  
6.         String subject = "TOOL.DEFAULT";  
7.         System.out.println(user +" "\+ password+" "\+url+" "\+subject+" ");  
8.           
9.         ConnectionFactory contectionFactory = new ActiveMQConnectionFactory(user,password,url);  
10.         try \{  
11.             Connection connection = contectionFactory.createConnection();  
12.             connection.start();  
13.             Session session = connection.createSession(Boolean.TRUE, Session.AUTO\_ACKNOWLEDGE);  
14.             Destination destination = session.createQueue(subject);  
15.             MessageProducer producer = session.createProducer(destination);  
16.             for(int i = 0;i<=20;i++)\{  
17.                 MapMessage message = session.createMapMessage();  
18.                 message.setLong("date", new Date().getTime());  
19.                 Thread.sleep(5000);  
20.                 producer.setDeliveryMode(DeliveryMode.NON\_PERSISTENT);  
21.                 producer.send(message);  
22.                   
23.                 System.out.println("—SendMessge："\+new Date());  
24.             \}  
25.             session.commit();  
26.             session.close();  
27.             connection.close();  
28.         \} catch (JMSException e) \{  
29.             e.printStackTrace();  
30.         \} catch (InterruptedException e) \{  
31.             e.printStackTrace();  
32.         \}  
33.     \}  
34. \}

注意：producer.setDeliveryMode(DeliveryMode.*NON\_PERSISTENT*);

此处显式指明DeliveryMode为NON\_PERSISENT

Java代码   ![收藏代码][icon_star.png]

1.  public class Consumer \{  
2.    
3.      public static void main(String\[\] args) \{  
4.          String user = ActiveMQConnection.DEFAULT\_USER;  
5.          String password = ActiveMQConnection.DEFAULT\_PASSWORD;  
6.          String url = ActiveMQConnection.DEFAULT\_BROKER\_URL;  
7.          String subject = "TOOL.DEFAULT";  
8.          ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user,password,url);  
9.          Connection connection;  
10.         try \{  
11.             connection = connectionFactory.createConnection();  
12.             connection.start();  
13.             final Session session = connection.createSession(Boolean.TRUE, Session.AUTO\_ACKNOWLEDGE);  
14.             Destination destination = session.createQueue(subject);  
15.             MessageConsumer message = session.createConsumer(destination);  
16.             message.setMessageListener(new MessageListener()\{  
17.   
18.                 public void onMessage(Message msg) \{  
19.                     MapMessage message = (MapMessage)msg;  
20.                     try \{  
21.                         System.out.println("--Receive："\+new Date(message.getLong("date")));  
22.                         session.commit();  
23.                     \} catch (JMSException e) \{  
24.                         e.printStackTrace();  
25.                     \}  
26.                 \}  
27.             \});  
28.             Thread.sleep(30000);  
29.             session.close();  
30.             connection.close();  
31.         \} catch (JMSException e) \{  
32.             e.printStackTrace();  
33.         \} catch (InterruptedException e) \{  
34.             e.printStackTrace();  
35.         \}  
36.     \}  
37.   
38. \}

ActiveMq Broker的配置都为默认。

先启动Broker,再启动Producer，等待Producer的20条Message发送完毕，Prducer程序运行结束，之后再启动Consumer,20条Message正常接收。

可见PERSISTENT与NON\_PERSISTENT,并不是指Broker会不会在Consumer未连接的情况为其存储Message

实验二：

先启动Broker,再启动Producer，等待Producer的20条Message发送完毕，Prducer程序运行结束。关闭ActiveMq  Broker服务，然后重启。之后再启动Consumer。

Consumer没有接收到任何消息。

实验三：

producer.setDeliveryMode(DeliveryMode.*PERSISTENT*);

在Producer中显式指明DeliveryMode为PERSISENT

实验步骤与实验二相同，但是此次Consumer收到了来自Producer的消息。

总结：

Persistent 用来指定JMS Provider对消息进行持久化操作，以免Provider fail的时候，丢失Message.

NON\_Persistent 方式下的JMS Provider不会对消进宪持久化，但上述实验一可知，Consumer还是会收到Message,可见JMS Provider会将相应的消息存在内存中，当Consumer连接上时，再发送过去，但在Provider fail的时候，Message会丢失。

事实上ActiveMq提供了多种消息持久化方式，包括AMQ、KahaDB、JDBC、LevelDB。从5.4版本之后KahaDB做为默认的持久化方式。

以下网上摘的：

消息订阅分为非持久订阅(non-durable subscription)和持久订阅(durablesubscription)，非持久订阅只有当客户端处于激活状态，也就是和JMS Provider保持连接状态才能收到发送到某个主题的消息，而当客户端处于离线状态，这个时间段发到主题的消息将会丢失，永远不会收到。持久订阅时，客户端向JMS注册一个识别自己身份的ID，当这个客户端处于离线时，JMS Provider 会为这个ID 保存所有发送到主题的消息，当客户再次连接到JMS Provider时，会根据自己的ID得到所有当自己处于离线时发送到主题的消息。

Topic 主题由JMS Provider 管理，主题由主题名识别，客户端可以通过JNDI接口用主题名得到一个主题对象。

<table style="color:rgb(0,0,0)"> 
 <tbody> 
 <tr> 
 <td> 消息发送端 </td> 
 <td> 消息接收端 </td> 
 <td> 可靠性及因素 </td> 
 </tr> 
 <tr> 
 <td> PERSISTENT </td> 
 <td> queue receiver/durable subscriber </td> 
 <td> 消费一次且仅消费一次。可靠性最好，但是占用服务器资源比较多。 </td> 
 </tr> 
 <tr> 
 <td> PERSISTENT </td> 
 <td> non-durable subscriber </td> 
 <td> 最多消费一次。这是由于non-durable subscriber决定的，如果消费端宕机或其他问题导致与JMS服务器断开连接，等下次再联上JMS服务器时的一系列消息，不为之保留。 </td> 
 </tr> 
 <tr> 
 <td> NON_PERSISTENT </td> 
 <td> queue receiver/durable subscriber </td> 
 <td> 最多消费一次。这是由于服务器的宕机会造成消息丢失 </td> 
 </tr> 
 <tr> 
 <td> NON_PERSISTENT </td> 
 <td> non-durable subscriber </td> 
 <td> 最多消费一次。这是由于服务器的宕机造成消息丢失，也可能是由于non-durable subscriber的性质所决定 
 <div> 
 
 </div> </td> 
 </tr> 
 </tbody> 
</table>

[icon_star.png]: /images/20220711/a6a3c832e96349f385d86da36538b69a.png