2.8、ActiveMQ 原生模式下 Topic 的持久化订阅

╰+哭是因爲堅強的太久メ 2021-12-12 06:37 413阅读 0赞

### 文章目录 ###

*   *   *  前言
         *  ActiveMQ 模式会持久化保存消息
         *  持久化保存不同于持久化订阅
         *  durable(持久化) 订阅模式
         *  Topic 持久化订阅
         *  代码
         *   *  生产者的关键配置——对消息持久化
             *  消费者关键配置——增加一个身份标识
             *   *  只有 Topic 的持久化订阅
                 *  代码
             *  管理页面
             *   *  只有 Topic 的持久化订阅
                 *  管理页面图
         *  参考链接

### 前言 ###

> 体能状态先于精神状态，习惯先于决心，聚焦先于喜好。

### ActiveMQ 模式会持久化保存消息 ###

> ActiveMQ 有很多消息持久化策略,其中KahaDB是默认的持久化策略，所有消息顺序添加到一个日志文件中，同时另外有一个索引文件记录指向这些日志的存储地址，还有一个事务日志用于消息回复操作。

### 持久化保存不同于持久化订阅 ###

> 但是这不意味着持久化订阅,比如ActiveMQ对Topic类型对订阅就是不可靠的,即便消费者没有确认消息,或者消费者当时不在线,都不会触发二次消息投递.

### durable(持久化) 订阅模式 ###

> 在 ActiveMQ 中，Queue 默认是持久化的，但是Topic 是非持久化的。  
> 如果我们想让 Topic 变成持久化，可以通过一些操作来实现。

### Topic 持久化订阅 ###

> Topic 进行持久化，相当于每一个消费者需要一个身份标识 clientID，以这个身份标识来订阅一个主题，然后需要制定一个新的名字，ActiveMQ会将主题的消息备份一个给新名字下的队列，当这个持久化订阅被感知时，ActiveMQ 会确保之后的该主题下的消息都备份一条，当消费者在线，则发送——需要确认机制，如果消费者不在线，则先保留，等到消费者在线时可以再次获取。  
> Topic 允许多个订阅，所以不同客户端可以以不同的clientID 持久化订阅这个Topic.

### 代码 ###

#### 生产者的关键配置——对消息持久化 ####

> 当消费者使用持久化订阅时,生产者端默认就会是这个模式  
> producer.setDeliveryMode(DeliveryMode.PERSISTENT);  
> 尽管从一般来说,仅需消费者端设置相应端持久化订阅即可,即如果你没有显式的声明持久化设置,对于可持久化的消息这是一个默认选项,即默认ActiveMQ会将生产者的消息保存到日志文件(默认 KahaDB)或者数据库中.  
> 但是某种情况下你还是可能会遇到收不到消息的情况 ([http://activemq.apache.org/why-do-i-not-receive-messages-on-my-durable-topic-subscription][http_activemq.apache.org_why-do-i-not-receive-messages-on-my-durable-topic-subscription])此时你还是需要在生产者端显式到声明一下 producer.setDeliveryMode(DeliveryMode.PERSISTENT);

import javax.jms.Connection;
    import javax.jms.DeliveryMode;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    /** * 生产者端使用 持久化模式 * @author jie.wu */
    public class ActiveMQDurableTopicSendTest { 
    	//发送方将消息设置为 persist
    	
    	@Test
    	public void durableTopicSend() { 
    		ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory();
    		Connection connection=null;
    		Session session=null;
    		
    		try { 
    			//获得 ActiveMQ 连接
    			connection=factory.createConnection();
    			//启动连接
    			connection.start();
    			//获得 Session
    			session=connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
    			//创建 Topic
    			Topic topic=session.createTopic("topic001");
    			//创建生产者
    			MessageProducer producer=session.createProducer(null);
    			producer.setDeliveryMode(DeliveryMode.PERSISTENT);
    			String msg="你好呀,我是一个消息topic001:"+System.currentTimeMillis();
    			//创建消息
    			TextMessage message=session.createTextMessage(msg);
    			System.out.println("生产者:"+msg);
    			//向消息队列发送消息
    			producer.send(topic,message);
    			//提交事务
    			session.commit();
    			
    		} catch (JMSException e) { 
    			System.out.println("出现问题:"+e);
    		}finally { 
    				try { 
    					if(connection!=null) { 
    						connection.stop();
    						connection.close();
    					}
    					if(session!=null) { 
    						session.close();
    					}
    				} catch (JMSException e) { 
    					System.out.println("关闭连接出现问题:"+e);
    				}
    			}
    	}
    }

#### 消费者关键配置——增加一个身份标识 ####

##### 只有 Topic 的持久化订阅 #####

> MessageConsumer consumer=session.createDurableSubscriber(topic, “durable001”);  
> 从 createDurableSubscriber 的方法可以看出,只支持 Topic 的持久化订阅

TopicSubscriber createDurableSubscriber(Topic topic, String name)
            throws JMSException;

##### 代码 #####

import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    /** * 消费者端使用 持久化模式 * @author jie.wu */
    public class ActiveMQDurableTopicReceiveTest { 
    	
    	@Test
    	public void durableTopicReceive() { 
    		
    		ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
    		Connection connection=null;
    		Session session=null;
    		
    		try { 
    			//创建连接
    			connection=factory.createConnection();
    			//指定客户端ID,ActiveMQ 会以 clientID 来识别不同的客户端
    			connection.setClientID("client001");
    			//启动连接
    			connection.start();
    			//创建session
    			session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			//创建 Topic
    			Topic topic=session.createTopic("topic001");
    			
    			//为 topic 创建一个 持久化 订阅,新名字为topic001-durable
    			//结合上面,消费者 client001 为主题 topic001 创建了一个持久化主题订阅 topic001-durable 
    			MessageConsumer consumer=session.createDurableSubscriber(topic, "durable001");
    			//消费者从ActiveMQ 获取一条信息
    			TextMessage text=(TextMessage) consumer.receive();
    			System.out.println("消费者:"+text.getText());
    			//需要手动确认消息被成功处理
    			text.acknowledge();
    			
    		} catch (JMSException e) { 
    			System.out.println("ActiveMQ 报错了:"+e);
    		}finally { 
    			try { 
    				if(connection!=null) { 
    					connection.stop();
    					connection.close();
    				}
    			} catch (JMSException e) { 
    				System.out.println("关闭资源报错了:"+e);
    			}
    		}
    	}
    }

#### 管理页面 ####

##### 只有 Topic 的持久化订阅 #####

> 只有 Topic 有持久化订阅的概念,Queue没有,因为持久化订阅的效果在消息特性上等同于 Queue  
> 从 ActiveMQ 提供的管理页面也可见一般

##### 管理页面图 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jlc3RjeHg_size_16_color_FFFFFF_t_70]

### 参考链接 ###

> \[1\]、[https://blog.csdn.net/qq\_20597727/article/details/81320298][https_blog.csdn.net_qq_20597727_article_details_81320298]  
> \[2\]、[http://activemq.apache.org/how-do-i-make-messages-durable][http_activemq.apache.org_how-do-i-make-messages-durable]  
> \[3\]、[http://activemq.apache.org/why-do-i-not-receive-messages-on-my-durable-topic-subscription][http_activemq.apache.org_why-do-i-not-receive-messages-on-my-durable-topic-subscription]

[http_activemq.apache.org_why-do-i-not-receive-messages-on-my-durable-topic-subscription]: http://activemq.apache.org/why-do-i-not-receive-messages-on-my-durable-topic-subscription
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jlc3RjeHg_size_16_color_FFFFFF_t_70]: /images/20211211/26c22497bdf9483789499a384cb1b577.png
[https_blog.csdn.net_qq_20597727_article_details_81320298]: https://blog.csdn.net/qq_20597727/article/details/81320298
[http_activemq.apache.org_how-do-i-make-messages-durable]: http://activemq.apache.org/how-do-i-make-messages-durable