AMQ学习笔记 - 16. 确认机制的测试
2024-05-12 23:47:36
概述
对Acknowledge机制进行测试。
此处的测试是针对Consumer的确认设计的；对于Producer的确认是透明的，无法提供测试。
测试实例
设计demo，测试三种确认机制。
| 测试机制 | 测试实例 | 结果预测 |
| AUTO_ACKNOWLEDGE | 接收正常 | 消息出队量=消息入队量 |
| 接收异常 | 消息出队量=0 | |
| CLIENT_ACKNOWLEDGE | 1次确认/2条消息 - 每2条消息确认1次 | 每次确认2条信息 |
| 从不确认 | 消息出队量=0 | |
| DUPS_OK_ACKNOWLEDGE | 每一次接收消息后，使线程睡眠数秒；观察消息出队情况 | 符合批量确认、延迟确认的特点 |
demo设计
demo设计图
测试分工
| 测试类 | 测试方法 |
|
AutoAckConsumer.java - 测试AUTO_ACKNOWLEDGE |
receiveNormal():void - 测试“接收正常” |
|
receiveIntentionalException():void - 测试“接收异常” |
|
|
ClientAckConsumer.java - 测试CLIENT_ACKNOWLEDGE |
receivePerTwice():void - 测试“1次确认/2条消息” |
|
receiveWithoutAck():void - 测试“从不确认” |
|
|
DupsOkAckConsumer.java - 测试DUPS_OK_ACKNOWLEDGE |
receive():void - 测试批量确认和延迟确认 |
测试步骤和结果
1.测试AUTO_ACKNOWLEDGE
1.1.接收正常
| 测试步骤 |
|
| 测试截图 |
|
|
1.2.接收异常
| 测试步骤 |
|
| 测试截图 |
|
|
| 结论整理 |
|
2.测试CLIENT_ACKNOWLEDGE
2.1.每2条消息确认1次
| 测试步骤 |
|
| 测试截图 |
|
|
| 结论整理 |
| 每次确认不是只对当前的Message进行确认，而是对自上次确认以来的所有Message进行确认.在这里，每次确认2条. |
2.2.从不确认
| 测试步骤 |
|
| 测试截图 |
|
|
3.测试DUPS_OK_ACKNOWLEDGE
| 测试步骤 |
|
| 结论整理 |
|
代码
文件目录结构
1 jms-producer 2 |---- src/main/resources/ 3 |---- jndi.properties 4 |---- src/main/java/ 5 |---- cn.sinobest.asj.producer.jms.acknowledge 6 |---- SimpleProducer.java # 发送 7 jms-consumer 8 |---- src/main/resources/ 9 |---- jndi.properties 10 |---- src/main/java/ 11 |---- cn.sinobest.asj.consumer.jms.acknowledge 12 |---- AutoAckConsumer.java # 测试AUTO_ACKNOWLEDGE 13 |---- ClientAckConsumer.java # 测试AUTO_ACKNOWLEDGE 14 |---- DupsOkAckConsumer.java # 测试DUPS_OK_ACKNOWLEDGE
文件内容
1.jndi.properties
jms-producer端
1 java.naming.factory.initial=org.apache.activemq.jndi.ActiveMQInitialContextFactory 2 3 # use the following property to configure the default connector 4 java.naming.provider.url=tcp://localhost:61616 5 6 # register some queues in JNDI using the form 7 # queue.[jndiName] = [physicalName] 8 queue.exampleQueue=example.queue 9 10 # register some topics in JNDI using the form 11 # topic.[jndiName] = [physicalName] 12 topic.exampleTopic=example.topic
jms-consumer端
1 java.naming.factory.initial=org.apache.activemq.jndi.ActiveMQInitialContextFactory 2 3 # use the following property to configure the default connector 4 java.naming.provider.url=tcp://localhost:61616 5 6 # register some queues in JNDI using the form 7 # queue.[jndiName] = [physicalName] 8 queue.exampleQueue=example.queue 9 10 # register some topics in JNDI using the form 11 # topic.[jndiName] = [physicalName] 12 topic.exampleTopic=example.topic
2.SimpleProducer.java
1 package cn.sinobest.asj.producer.jms.acknowledge;
2 import javax.jms.Connection;
3 import javax.jms.ConnectionFactory;
4 import javax.jms.Destination;
5 import javax.jms.JMSException;
6 import javax.jms.MessageProducer;
7 import javax.jms.Session;
8 import javax.jms.TextMessage;
9 import javax.naming.Context;
10 import javax.naming.InitialContext;
11 import javax.naming.NamingException;
12 import org.junit.Test;
13 /**
14 * A simple demo for producer client to send message to ActiveMQ.<br>
15 * 对{@link cn.sinobest.asj.producer.jms.clientmode.SimpleProducer}的改进.
16 *
17 * @author lijinlong
18 *
19 */
20 public class SimpleProducer {
21 /** JNDI name for ConnectionFactory */
22 static final String CONNECTION_FACTORY_JNDI_NAME = "ConnectionFactory";
23 /** JNDI name for Queue Destination (use for PTP Mode) */
24 static final String QUEUE_JNDI_NAME = "exampleQueue";
25 /** JNDI name for Topic Destination (use for Pub/Sub Mode) */
26 static final String TOPIC_JNDI_NAME = "exampleTopic";
27 /**
28 * 发送消息到队列.<br>
29 * PTP Mode.
30 */
31 @Test
32 public void sendToQueue() {
33 send(QUEUE_JNDI_NAME);
34 }
35
36 /**
37 * 发送消息到主题.<br>
38 * PTP Mode.
39 */
40 @Test
41 public void sendToTopic() {
42 send(TOPIC_JNDI_NAME);
43 }
44 /**
45 * 发送到指定的目的地.
46 *
47 * @param destJndiName
48 * 目的地的JNDI name：{@link #QUEUE_JNDI_NAME}或
49 * {@link #TOPIC_JNDI_NAME}.
50 */
51 private void send(String destJndiName) {
52 Context jndiContext = null;
53 ConnectionFactory connectionFactory = null;
54 Connection connection = null;
55 Session session = null;
56 Destination destination = null;
57 MessageProducer producer = null;
58 // create a JNDI API IntialContext object
59 try {
60 jndiContext = new InitialContext();
61 } catch (NamingException e) {
62 System.out.println("Could not create JNDI Context:"
63 + e.getMessage());
64 System.exit(1);
65 }
66 // look up ConnectionFactory and Destination
67 try {
68 connectionFactory = (ConnectionFactory) jndiContext
69 .lookup(CONNECTION_FACTORY_JNDI_NAME);
70 destination = (Destination) jndiContext.lookup(destJndiName);
71 } catch (NamingException e) {
72 System.out.println("JNDI look up failed:" + e.getMessage());
73 System.exit(1);
74 }
75 // send Messages and finally release the resources.
76 try {
77 connection = connectionFactory.createConnection();
78 session = connection.createSession(Boolean.FALSE,
79 Session.AUTO_ACKNOWLEDGE);
80 producer = session.createProducer(destination);
81 TextMessage message = session.createTextMessage();
82 for (int i = 0; i < 3; i++) {
83 message.setText(String.format("This is the %dth message.",
84 i + 1));
85 producer.send(message);
86 }
87 } catch (JMSException e) {
88 e.printStackTrace();
89 } finally {
90 try {
91 if (session != null)
92 session.close();
93 if (connection != null)
94 connection.close();
95 } catch (JMSException e) {
96 e.printStackTrace();
97 }
98 }
99 }
100 }SimpleProducer.java
3.AutoAckConsumer.java
1 package cn.sinobest.asj.consumer.jms.acknowledge;
2 import javax.jms.Connection;
3 import javax.jms.ConnectionFactory;
4 import javax.jms.Destination;
5 import javax.jms.JMSException;
6 import javax.jms.Message;
7 import javax.jms.MessageConsumer;
8 import javax.jms.MessageListener;
9 import javax.jms.Session;
10 import javax.jms.TextMessage;
11 import javax.naming.Context;
12 import javax.naming.InitialContext;
13 import javax.naming.NamingException;
14 import org.junit.Test;
15 import cn.sinobest.asj.consumer.util.Hold;
16 /**
17 * AUTO_ACKNOWLEDGE确认模式的Consumer.<br>
18 * 基于PTP Mode，采用异步的方式接收消息，研究抛出或不抛出异常的情况下，Queue中的消息的出队情况.<br>
19 *
20 * @author lijinlong
21 *
22 */
23 public class AutoAckConsumer {
24 /** JNDI name for ConnectionFactory */
25 static final String CONNECTION_FACTORY_JNDI_NAME = "ConnectionFactory";
26 /** JNDI name for Queue Destination (use for PTP Mode) */
27 static final String QUEUE_JNDI_NAME = "exampleQueue";
28 /**
29 * 正常的接收.<br>
30 */
31 @Test
32 public void receiveNormal() {
33 MessageListener listener = new MessageListener() {
34 public void onMessage(Message message) {
35 try {
36 String text = ((TextMessage) message).getText();
37 System.out.println(text);
38 } catch (JMSException e) {
39 e.printStackTrace();
40 }
41 }
42 };
43 receive(listener);
44 }
45 /**
46 * 故意抛出异常的接收.<br>
47 * 结果：
48 * <ul>
49 * <li>JMS Provider重复发送消息给Consumer。重复次数达到一定的阀值，JMS
50 * Provider认为此消息无法消费，此消息将会被删除或者迁移到"dead letter"通道中。</li>
51 * <li>在测试过程中，会重发6次（共发7次），然后移到ActiveMQ.DLQ队列；DLQ - dead letter queue.</li>
52 * <li>重发次数可以配置 -
53 * 在brokerUrl中指定参数jms.redeliveryPolicy.maximumRedeliveries=3，则重发3次（共4次）.</li>
54 * </ul>
55 */
56 @Test
57 public void receiveIntentionalException() {
58 MessageListener listener = new MessageListener() {
59 public void onMessage(Message message) {
60 try {
61 String text = ((TextMessage) message).getText();
62 System.out.println(text);
63 } catch (JMSException e) {
64 e.printStackTrace();
65 }
66 boolean intentional = true;
67 if (intentional) {
68 throw new RuntimeException("故意抛出的异常。");
69 }
70 }
71 };
72 receive(listener);
73 }
74
75 /**
76 * 接收消息.<br>
77 *
78 * @param listener
79 * 监听器，如果消息接收成功，将被回调.
80 */
81 private void receive(MessageListener listener) {
82 Context jndiContext = null;
83 ConnectionFactory connectionFactory = null;
84 Connection connection = null;
85 Session session = null;
86 Destination destination = null;
87 MessageConsumer consumer = null;
88 // create a JNDI API IntialContext object
89 try {
90 jndiContext = new InitialContext();
91 } catch (NamingException e) {
92 System.out.println("Could not create JNDI Context:"
93 + e.getMessage());
94 System.exit(1);
95 }
96 // look up ConnectionFactory and Destination
97 try {
98 connectionFactory = (ConnectionFactory) jndiContext
99 .lookup(CONNECTION_FACTORY_JNDI_NAME);
100 destination = (Destination) jndiContext.lookup(QUEUE_JNDI_NAME);
101 } catch (NamingException e) {
102 System.out.println("JNDI look up failed:" + e.getMessage());
103 System.exit(1);
104 }
105 // receive Messages and finally release the resources.
106 try {
107 connection = connectionFactory.createConnection();
108 connection.start(); // connection should be called in
109 // receiver-client
110 session = connection.createSession(Boolean.FALSE,
111 Session.AUTO_ACKNOWLEDGE);
112 consumer = session.createConsumer(destination);
113 // key code for asynchronous receive:set messageListener
114 consumer.setMessageListener(listener);
115 Hold.hold(); // 阻塞程序继续执行
116 } catch (JMSException e) {
117 e.printStackTrace();
118 } finally {
119 try {
120 if (session != null)
121 session.close();
122 if (connection != null)
123 connection.close();
124 } catch (JMSException e) {
125 e.printStackTrace();
126 }
127 }
128 }
129 }AutoAckConsumer.java
4.ClientAckConsumer.java
1 package cn.sinobest.asj.consumer.jms.acknowledge;
2 import javax.jms.Connection;
3 import javax.jms.ConnectionFactory;
4 import javax.jms.Destination;
5 import javax.jms.JMSException;
6 import javax.jms.Message;
7 import javax.jms.MessageConsumer;
8 import javax.jms.MessageListener;
9 import javax.jms.Session;
10 import javax.jms.TextMessage;
11 import javax.naming.Context;
12 import javax.naming.InitialContext;
13 import javax.naming.NamingException;
14 import org.junit.Test;
15 import cn.sinobest.asj.consumer.util.Hold;
16 /**
17 * CLIENT_ACKNOWLEDGE确认模式的Consumer.<br>
18 * 基于PTP Mode，采用异步的方式接收消息，研究从不确认、每2次确认的情况下，Queue中的消息的出队情况.<br>
19 *
20 * @author lijinlong
21 *
22 */
23 public class ClientAckConsumer {
24 /** JNDI name for ConnectionFactory */
25 static final String CONNECTION_FACTORY_JNDI_NAME = "ConnectionFactory";
26 /** JNDI name for Queue Destination (use for PTP Mode) */
27 static final String QUEUE_JNDI_NAME = "exampleQueue";
28 /**
29 * 从不确认的接收.<br>
30 * 结果：
31 * <ul>
32 * <li>只接收一次，但是消息不会出队.</li>
33 * <li>Consumer重启，会再次接收到消息.</li>
34 * </ul>
35 */
36 @Test
37 public void receiveWithoutAck() {
38 MessageListener listener = new MessageListener() {
39 public void onMessage(Message message) {
40 try {
41 String text = ((TextMessage) message).getText();
42 System.out.println(text);
43 } catch (JMSException e) {
44 e.printStackTrace();
45 }
46 }
47 };
48 receive(listener);
49 }
50
51 private int ack_count = 0; // 确认次数统计
52 /**
53 * 每接收两次确认一次.<br>
54 * 结果：每次确认不是只对当前的Message进行确认，而是对自上次确认以来的所有Message进行确认.在这里，每次确认2条.
55 */
56 @Test
57 public void receivePerTwice() {
58 MessageListener listener = new MessageListener() {
59 public void onMessage(Message message) {
60 try {
61 String text = ((TextMessage) message).getText();
62 System.out.println(text);
63
64 ack_count ++;
65 if (ack_count % 2 == 0)
66 message.acknowledge();
67
68 } catch (JMSException e) {
69 e.printStackTrace();
70 }
71 }
72 };
73 receive(listener);
74 }
75 /**
76 * 接收消息.<br>
77 *
78 * @param listener
79 * 监听器，如果消息接收成功，将被回调.
80 */
81 private void receive(MessageListener listener) {
82 Context jndiContext = null;
83 ConnectionFactory connectionFactory = null;
84 Connection connection = null;
85 Session session = null;
86 Destination destination = null;
87 MessageConsumer consumer = null;
88 // create a JNDI API IntialContext object
89 try {
90 jndiContext = new InitialContext();
91 } catch (NamingException e) {
92 System.out.println("Could not create JNDI Context:"
93 + e.getMessage());
94 System.exit(1);
95 }
96 // look up ConnectionFactory and Destination
97 try {
98 connectionFactory = (ConnectionFactory) jndiContext
99 .lookup(CONNECTION_FACTORY_JNDI_NAME);
100 destination = (Destination) jndiContext.lookup(QUEUE_JNDI_NAME);
101 } catch (NamingException e) {
102 System.out.println("JNDI look up failed:" + e.getMessage());
103 System.exit(1);
104 }
105 // receive Messages and finally release the resources.
106 try {
107 connection = connectionFactory.createConnection();
108 connection.start(); // connection should be called in
109 // receiver-client
110 session = connection.createSession(Boolean.FALSE,
111 Session.CLIENT_ACKNOWLEDGE);
112 consumer = session.createConsumer(destination);
113 // key code for asynchronous receive:set messageListener
114 consumer.setMessageListener(listener);
115 Hold.hold(); // 阻塞程序继续执行
116 } catch (JMSException e) {
117 e.printStackTrace();
118 } finally {
119 try {
120 if (session != null)
121 session.close();
122 if (connection != null)
123 connection.close();
124 } catch (JMSException e) {
125 e.printStackTrace();
126 }
127 }
128 }
129 }ClientAckConsumer.java
5.DupsOkAckConsumer.java
1 package cn.sinobest.asj.consumer.jms.acknowledge;
2 import javax.jms.Connection;
3 import javax.jms.ConnectionFactory;
4 import javax.jms.Destination;
5 import javax.jms.JMSException;
6 import javax.jms.Message;
7 import javax.jms.MessageConsumer;
8 import javax.jms.MessageListener;
9 import javax.jms.Session;
10 import javax.jms.TextMessage;
11 import javax.naming.Context;
12 import javax.naming.InitialContext;
13 import javax.naming.NamingException;
14 import org.junit.Test;
15 import cn.sinobest.asj.consumer.util.Hold;
16 /**
17 * DUPS_OK_ACKNOWLEDGE确认模式的Consumer.<br>
18 * @author lijinlong
19 *
20 */
21 public class DupsOkAckConsumer {
22 /** JNDI name for ConnectionFactory */
23 static final String CONNECTION_FACTORY_JNDI_NAME = "ConnectionFactory";
24 /** JNDI name for Topic Destination (use for Pub/Sub Mode) */
25 static final String TOPIC_JNDI_NAME = "exampleTopic";
26
27 /**
28 * 从主题接收消息.
29 */
30 @Test
31 public void receive() {
32 receive(createMessageListener());
33 }
34
35 /**
36 * 创建MessageListener实例.
37 * @return
38 */
39 private MessageListener createMessageListener() {
40 MessageListener listener = new MessageListener() {
41 public void onMessage(Message message) {
42 try {
43 String text = ((TextMessage) message).getText();
44 System.out.println(text);
45 } catch (JMSException e) {
46 e.printStackTrace();
47 }
48
49 try {
50 Thread.sleep(5 * 1000);
51 } catch (InterruptedException e) {
52 e.printStackTrace();
53 }
54 }
55 };
56
57 return listener;
58 }
59
60 /**
61 * 接收消息.<br>
62 *
63 * @param listener
64 * 监听器，如果消息接收成功，将被回调.
65 */
66 private void receive(MessageListener listener) {
67 Context jndiContext = null;
68 ConnectionFactory connectionFactory = null;
69 Connection connection = null;
70 Session session = null;
71 Destination destination = null;
72 MessageConsumer consumer = null;
73 // create a JNDI API IntialContext object
74 try {
75 jndiContext = new InitialContext();
76 } catch (NamingException e) {
77 System.out.println("Could not create JNDI Context:"
78 + e.getMessage());
79 System.exit(1);
80 }
81 // look up ConnectionFactory and Destination
82 try {
83 connectionFactory = (ConnectionFactory) jndiContext
84 .lookup(CONNECTION_FACTORY_JNDI_NAME);
85 destination = (Destination) jndiContext.lookup(TOPIC_JNDI_NAME);
86 } catch (NamingException e) {
87 System.out.println("JNDI look up failed:" + e.getMessage());
88 System.exit(1);
89 }
90 // receive Messages and finally release the resources.
91 try {
92 connection = connectionFactory.createConnection();
93 connection.start(); // connection should be called in
94 // receiver-client
95 session = connection.createSession(Boolean.FALSE,
96 Session.DUPS_OK_ACKNOWLEDGE);
97 consumer = session.createConsumer(destination);
98 // key code for asynchronous receive:set messageListener
99 consumer.setMessageListener(listener);
100 Hold.hold(); // 阻塞程序继续执行
101 } catch (JMSException e) {
102 e.printStackTrace();
103 } finally {
104 try {
105 if (session != null)
106 session.close();
107 if (connection != null)
108 connection.close();
109 } catch (JMSException e) {
110 e.printStackTrace();
111 }
112 }
113 }
114 }DupsOkAckConsumer.java
来自为知笔记(Wiz)
转载于:https://www.cnblogs.com/ywjy/articles/5434776.html
AMQ学习笔记 - 16. 确认机制的测试相关推荐
- Linux 学习笔记16 信号量
Linux 学习笔记16 信号量Semaphore 信号量概念 信号量(或信号灯)是一种用于提供不同进程间或一个给定进程的不同线程间同步手段的原语. 信号量是控制进程(或线程)同步(谁先执行,谁后执行 ...
- SpringBoot学习笔记(16)----SpringBoot整合Swagger2
Swagger 是一个规范和完整的框架,用于生成,描述,调用和可视化RESTful风格的web服务 http://swagger.io Springfox的前身是swagger-springmvc,是 ...
- Netty网络框架学习笔记-16(心跳(heartbeat)服务源码分析)
Netty网络框架学习笔记-16(心跳(heartbeat)服务源码分析_2020.06.25) 前言: Netty 作为一个网络框架,提供了诸多功能,比如编码解码等,Netty 还提供了非常重要的一 ...
- 区块链学习笔记16——ETH交易树和收据树
区块链学习笔记16--ETH交易树和收据树 学习视频:北京大学肖臻老师<区块链技术与应用> 笔记参考:北京大学肖臻老师<区块链技术与应用>公开课系列笔记--目录导航页 交易树和 ...
- cocos2d-x学习笔记16：记录存储1：CCUserDefault
cocos2d-x学习笔记16:记录存储1:CCUserDefault 一.简述 CCUserDefalt作为NSUserDefalt类的cocos2d-x实现版本,承担了cocos2d-x引擎的记录 ...
- Hadoop学习笔记—16.Pig框架学习
Hadoop学习笔记-16.Pig框架学习 一.关于Pig:别以为猪不能干活 1.1 Pig的简介 Pig是一个基于Hadoop的大规模数据分析平台,它提供的SQL-LIKE语言叫Pig Latin, ...
- 台大李宏毅Machine Learning 2017Fall学习笔记 (16)Unsupervised Learning:Neighbor Embedding
台大李宏毅Machine Learning 2017Fall学习笔记 (16)Unsupervised Learning:Neighbor Embedding
- 【动手深度学习-笔记】注意力机制（一）注意力机制框架
生物学中的注意力提示 非自主性提示: 在没有主观意识的干预下,眼睛会不自觉地注意到环境中比较突出和显眼的物体. 比如我们自然会注意到一堆黑球中的一个白球,马路上最酷的跑车等. 自主性提示: 在主观意识 ...
- 【论文学习笔记-16】立体匹配：360SD-net
[论文学习笔记-16]立体匹配:360SD-net Contribution RelatedWork Method Experiment 本文利用两张360°摄像机获得的球形图片进行立体匹配,与双目立 ...
- 【动手深度学习-笔记】注意力机制（四）自注意力、交叉注意力和位置编码
文章目录 自注意力(Self-Attention) 例子 Self-Attention vs Convolution Self-Attention vs RNN 交叉注意力(Cross Attenti ...
最新文章
- 【BZOJ5102】[POI2018]Prawnicy 堆
- linux访问网页元素,Linux_DOM和JAVASCRIPT访问页面上的元素，访问方法：getElementById() - phpStudy...
- Struts项目中，检测用户名是否被占用/查询账户名称是否被占用/查询账户名称是否已被注册/检查用户名是否被注册
- python3cookbook_python3-cookbook读书笔记（四）
- MEF董事、中国电信云计算中心主任赵慧玲：MEF第三类网络
- 解决矩池云使用中ssh链接的时候日志丢失
- C#读写三菱Fx PLC 使用Fx 串口协议 读写Fx3U设备
- HCNA学习的第二天
- 6步解决win7局域网内传输慢的问题
- java指纹识别+谷歌图片识别技术
- 网络安全知识竞赛选择题(1-30题)
- 如何制作unity艺术字体
- 洛谷题解P1428 小鱼比可爱
- 【活动时间调整】博客搬家，有礼相送
- 关于java的面试题_JAVA面试题100问第一部分
- Java - String是最基本的数据类型吗？
- 只要五分钟，让你成功接入Twitter的第三方登录
- 影视寒冬下的影视作品“质检报告”——电视剧篇
- 用Arduino读取HX711应变片专用模块
- 我的世界服务器物品属性,[娱乐|机械]Item Lore Stats —— 自定义你的物品属性[自带强化][1.10及以上]...