2.7、ActiveMQ 原生模式下 Queue 和 Topic 的 ACK 规律

女爷i 2021-12-11 22:49 223阅读 0赞

### 文章目录 ###

*   *   *  前言
         *  特别声明
         *  感受原生ActiveMQ中的 ack
         *  创建 session 时的两个入参
         *  连接ActiveMQ的一个小窍门
         *  生产者的 ack 规律
         *   *  生产者 session 开启事务
             *   *  代码
                 *  运行结果
                 *  在ActiveMQ 提供的管理后台查看
             *  生产者 session 不开启事务
             *   *  代码
                 *  运行结果
                 *  在ActiveMQ 提供的管理后台查看
         *  消费者的 ack 规律
         *   *  消费者 session 开启事务
             *   *  代码
                 *  在ActiveMQ 提供的管理后台查看
             *  消费者 session 不开启事务 ,手动 确认模式
             *   *  代码
                 *  运行结果
                 *  在ActiveMQ 提供的管理后台查看
             *  不开启事务及其他几种模式
             *  死信队列
         *  Topic 的 ack
         *   *  代码
             *  生产者发送而没有消费者
             *  生产者发送-消费者确认
             *  生产者发送-消费者未确认

### 前言 ###

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

### 特别声明 ###

> 注意,本文提供的代码不包含包路径,如需测试请自行补充.  
> 本文提供的对ActiveMQ ack机制的测试没有使用 Spring JMS template.

### 感受原生ActiveMQ中的 ack ###

> 在使用 ActiveMQ 原生代码进行消息的发送和接收室,我们可以使用默认的消息确认机制,也可以通过代码手动的确认消息的发送或者接收.第一点区别在于创建 session 的时候是否提交事务,第二点区别在于消息的确认机制.

### 创建 session 时的两个入参 ###

> 一个是是否开启事务,一个是消息确认模式,可以参考\[[https://blog.csdn.net/bestcxx/article/details/90551495][https_blog.csdn.net_bestcxx_article_details_90551495]\](ActiveMQ 的消息确认机制-ACK)  
> 对于服务端来说,只有第一个参数是有意义的,消息确认模式参数没有意义——但是对于客户端来说当事务不开启时是有意义的,否则也是没有意义的.——或许设计者是为了考虑服务端和客户端使用同一个session 对象?

Session createSession(boolean transacted, int acknowledgeMode)
            throws JMSException;

### 连接ActiveMQ的一个小窍门 ###

> 如果你在创建ActiveMQ的工厂和连接没有指定具体的ActiveMQ地址、用户名、密码的时候,ActiveMQ并不会报错.因为其内部封装了默认的连接配置.

//进去看下其封装的方法
    ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory();
    //进去看下其封装的方法
    Connection connection=factory.createConnection();

等同于下面的代码,在内部封装了默认参数

//指定一个默认的连接地址-其实就是本机
    ActiveMQConnectionFactory factory=
    new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
    //指定默认的用户和密码-其实都是null
    Connection connection=factory.createConnection(ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD);

### 生产者的 ack 规律 ###

#### 生产者 session 开启事务 ####

> session=connection.createSession(true, Session.AUTO\_ACKNOWLEDGE); 中true 即开启事务,第二个参数无效.  
> 开启事务的话,只有提交事务该消息才会被发送给 ActiveMQ.  
> session.commit(); 这一句就是提交事务.  
> 读者可以试试注释掉 session.commit(); 这一行代码,可以在ActiveMQ管理后台发现消息没有被ActiveMQ 接收.

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

import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    public class ActiveMQSendTest { 
    	
    	/** * 使用 ActiveMQ 发送消息 * 开启事务:需要提交事务,session.commit();否则消息不会被发送给ActiveMQ * 发送阶段消息确认设置没有实质作用 session=connection.createSession(true, Session.AUTO_ACKNOWLEDGE); */
    	@Test
    	public void testSendMessageWithTransacte() { 
    		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);
    			//创建 Queue
    			Queue queue=session.createQueue("test005");
    			//创建生产者
    			MessageProducer producer=session.createProducer(null);
    			//创建消息
    			TextMessage message=session.createTextMessage("你好呀,我是一个消息test001");
    			//向消息队列发送消息
    			producer.send(queue,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);
    				}
    			}
    	}
    }

##### 运行结果 #####

INFO | Successfully connected to tcp://localhost:61616

##### 在ActiveMQ 提供的管理后台查看 #####

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

#### 生产者 session 不开启事务 ####

> session=connection.createSession(false, Session.AUTO\_ACKNOWLEDGE); false即不开启事务,第二个参数对生产者一侧没有影响.  
> 不开启事务,只要运行了发送消息的代码,消息就会被发送给ActiveMQ;  
> 在生产者这边,虽然需要指定消息的确认机制,但是在消费者这端该配置不会影响消费者,除非二者公用一个 session 对象.

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

import javax.jms.Connection;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Queue;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    public class ActiveMQSendTest { 
    	/** * 使用 ActiveMQ 发送消息 * 不开启事务:无需提交事务 session.commit() * 发送阶段消息确认设置没有实质作用 session=connection.createSession(true, Session.AUTO_ACKNOWLEDGE); */
    	@Test
    	public void testSendMessageNoTransacte() { 
    		ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory();
    		Connection connection=null;
    		Session session=null;
    		try { 
    			connection=factory.createConnection();
    			session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			Queue queue=session.createQueue("test002");
    			MessageProducer producer=session.createProducer(null);
    			TextMessage message=session.createTextMessage("这是一条消息test002");
    			producer.send(queue,message);
    			//不用提交事务呦
    		} 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);
    			}
    		}
    	}
    }

##### 运行结果 #####

INFO | Successfully connected to tcp://localhost:61616

##### 在ActiveMQ 提供的管理后台查看 #####

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

### 消费者的 ack 规律 ###

> 在消费者端,开启事务或者不开启事务对消息确认的方式会有一些影响.  
> 具体来说,如果开启事务,第二个参数也会实效,只要事务提交即可,ActiveMQ会将客户端确认的消息清除队列  
> 但是如果没有

#### 消费者 session 开启事务 ####

> session=connection.createSession(true, Session.AUTO\_ACKNOWLEDGE); true开启事务  
> 开启事务必须提交事务,以让ActiveMQ知道消息被消费者成功接收了.  
> 此外,在客户端开启事务的时候,第二个参数也会失效.

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

import javax.jms.Connection;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    public class ActiveMQReceiveTest { 
    	
    	@Test
    	public void testReceiveWithTransacted() { 
    		ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
    		Connection connection=null;
    		Session session=null;
    		
    		try { 
    			//connection=factory.createConnection();
    			connection=factory.createConnection(ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD);
    			connection.start();
    	
    			//配置为开启事务但是不提交事务是错误配置: 
    			//可以重复接收 1+6=7 次数据,然后数据将进入 Active MQ的死信队列(Queue名称为ActiveMQ.DLQ)
    			session=connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
    			
    			Destination queue=session.createQueue("test005");
    			MessageConsumer consumer=session.createConsumer(queue);
    			TextMessage text=(TextMessage) consumer.receive();
    			System.out.println("text:"+text.getText());
    			session.commit();
    			
    		} catch (JMSException e) { 
    			System.out.println("ActiveMQ 报错了:"+e);
    		}finally { 
    			try { 
    				if(connection!=null) { 
    					connection.stop();
    					connection.close();
    				}
    			} catch (JMSException e) { 
    				System.out.println("关闭资源报错了:"+e);
    			}
    		}
    	}
    }

运行结果

INFO | Successfully connected to tcp://localhost:61616
    text:你好呀,我是一个消息test001

##### 在ActiveMQ 提供的管理后台查看 #####

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

#### 消费者 session 不开启事务 ,手动 确认模式 ####

> session=connection.createSession(false, Session.CLIENT\_ACKNOWLEDGE); false 即不开启事务,第二个参数将起作用.  
> 需要手动确认消息被成功处理,如果没有确认,测试可以无限获取.  
> textMessage.acknowledge();

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

import javax.jms.Connection;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    public class ActiveMQReceiveTest { 
    	@Test
    	public void testReceiveClientAcknowledgeNoTransacted() { 
    		ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
    		Connection connection=null;
    		Session session=null;
    		
    		try { 
    			//connection=factory.createConnection();
    			connection=factory.createConnection(ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD);
    			connection.start();
    			session=connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
    			
    			Destination queue=session.createQueue("test002");
    			MessageConsumer consumer=session.createConsumer(queue);
    			TextMessage text=(TextMessage) consumer.receive();
    			System.out.println("text:"+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);
    			}
    		}	
    	}
    }

##### 运行结果 #####

INFO | Successfully connected to tcp://localhost:61616
    text:这是一条消息test002

##### 在ActiveMQ 提供的管理后台查看 #####

![在这里插入图片描述][20190709192657295.png]

#### 不开启事务及其他几种模式 ####

> 参考 [ActiveMQ 的消息确认机制-ACK][https_blog.csdn.net_bestcxx_article_details_90551495],还有其他集中模式,在不启用事务时:AUTO\_ACKNOWLEDGE会自动确认;  
> DUPS\_OK\_ACKNOWLEDGE 会批量确认,即ActiveMQ 可能不会立即收到提示,客户端有可能可以重复消费同一条消息;  
> SESSION\_TRANSACTED 这个其实是开启事务时搭配的,你可以认为没啥用.  
> INDIVIDUAL\_ACKNOWLEDGE,每次确认一条,针对一个session消费多条确认一次来理解.

#### 死信队列 ####

> 如果消费者接收到 ActiveMQ 推送的消息,但是没有确认该消息,默认 ActiveMQ 允许客户端最多再次获取6次如果不成功的话,当额外次数达到6次,依旧不成功,这条消息就会被放入死信队列.  
> 读者可以使用 消费者开启事务,但是不提供事务这种方式连续获取信息进行尝试,最后会在 ActiveMQ 的控制台发现多了一个 队列 ActiveMQ.DLQ ——死信队列,如下

![在这里插入图片描述][20190709193222845.png]

### Topic 的 ack ###

> 当消费者不在线,而生产者向 对应Topic 发送消息,消费者启动后,不会得到之前的消息,在消费者在线,收到 消息后,使用ack机制不会造成 ActiveMQ 重复发送消息,或者说,消费者最多也只能收到一次消息,区别在于,如果消费者消费消息并确认,在监控页面是可以看到被消费的,否则情况等同于没有发送的情况.

#### 代码 ####

> 注意消费者确认消息的代码段

//需要手动确认消息被成功处理
    text.acknowledge();

*  生产者

import javax.jms.Connection;
    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;
    
    /** * 测试发送 Topic * @author jie.wu */
    public class ActiveMQTopicSendTest { 
    	
    	/** * 使用 ActiveMQ 发送消息 * 开启事务:需要提交事务,session.commit();否则消息不会被发送给ActiveMQ * 发送阶段消息确认设置没有实质作用 session=connection.createSession(true, Session.AUTO_ACKNOWLEDGE); */
    	@Test
    	public void testSendMessageWithTransacte() { 
    		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);
    			//创建消息
    			TextMessage message=session.createTextMessage("你好呀,我是一个消息topic001");
    			//向消息队列发送消息
    			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);
    				}
    			}
    	}
    
    }

*  消费者

import javax.jms.Connection;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageConsumer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.junit.Test;
    
    public class ActiveMQTopicReceiveTest { 
    	
    	@Test
    	public void testReceiveClientAcknowledgeNoTransacted() { 
    		ActiveMQConnectionFactory factory=new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
    		Connection connection=null;
    		Session session=null;
    		
    		try { 
    			//connection=factory.createConnection();
    			connection=factory.createConnection(ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD);
    			connection.start();
    			session=connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
    			
    			Destination queue=session.createTopic("topic001");
    			MessageConsumer consumer=session.createConsumer(queue);
    			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);
    			}
    		}
    		
    	}
    
    }

#### 生产者发送而没有消费者 ####

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

#### 生产者发送-消费者确认 ####

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

#### 生产者发送-消费者未确认 ####

> 消费者不开启事务,手动确认模式 session=connection.createSession(false, Session.CLIENT\_ACKNOWLEDGE);  
> 但是没有确认消息 text.acknowledge();

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

[https_blog.csdn.net_bestcxx_article_details_90551495]: https://blog.csdn.net/bestcxx/article/details/90551495
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jlc3RjeHg_size_16_color_FFFFFF_t_70]: /images/20211211/9225abaaa48d4ffaad6f50dd847d82ea.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jlc3RjeHg_size_16_color_FFFFFF_t_70 1]: /images/20211211/9e438a0aa54c4b0f8ab01b8bbf02e3f7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jlc3RjeHg_size_16_color_FFFFFF_t_70 2]: /images/20211211/78636dd008e04935ac67eb3685903c84.png
[20190709192657295.png]: /images/20211211/1861ee8994b84ab7953fcba31e6a3491.png
[20190709193222845.png]: /images/20211211/978f0d0abfd94cdc85d69152b97fc735.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jlc3RjeHg_size_16_color_FFFFFF_t_70 3]: /images/20211211/42becec31f6540ebb1cce0272d45c8e0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jlc3RjeHg_size_16_color_FFFFFF_t_70 4]: /images/20211211/90d1d121a58945e2b78e923c2b71d937.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jlc3RjeHg_size_16_color_FFFFFF_t_70 5]: /images/20211211/e1d4a035ee874da5aa085f852447a293.png