概述
对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
| 测试步骤 |
|
| 结论整理 |
|
代码
文件目录结构
jms-producer
|---- src/main/resources/
|---- jndi.properties
|---- src/main/java/
|---- cn.sinobest.asj.producer.jms.acknowledge
|---- SimpleProducer.java # 发送
jms-consumer
|---- src/main/resources/
|---- jndi.properties
|---- src/main/java/
|---- cn.sinobest.asj.consumer.jms.acknowledge
|---- AutoAckConsumer.java # 测试AUTO_ACKNOWLEDGE
|---- ClientAckConsumer.java # 测试AUTO_ACKNOWLEDGE
|---- DupsOkAckConsumer.java # 测试DUPS_OK_ACKNOWLEDGE
文件内容
1.jndi.properties
jms-producer端
java.naming.factory.initial=org.apache.activemq.jndi.ActiveMQInitialContextFactory # use the following property to configure the default connector
java.naming.provider.url=tcp://localhost:61616 # register some queues in JNDI using the form
# queue.[jndiName] = [physicalName]
queue.exampleQueue=example.queue # register some topics in JNDI using the form
# topic.[jndiName] = [physicalName]
topic.exampleTopic=example.topic
jms-consumer端
java.naming.factory.initial=org.apache.activemq.jndi.ActiveMQInitialContextFactory # use the following property to configure the default connector
java.naming.provider.url=tcp://localhost:61616 # register some queues in JNDI using the form
# queue.[jndiName] = [physicalName]
queue.exampleQueue=example.queue # register some topics in JNDI using the form
# topic.[jndiName] = [physicalName]
topic.exampleTopic=example.topic
2.SimpleProducer.java
package cn.sinobest.asj.producer.jms.acknowledge;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import org.junit.Test;
/**
* A simple demo for producer client to send message to ActiveMQ.<br>
* 对{@link cn.sinobest.asj.producer.jms.clientmode.SimpleProducer}的改进.
*
* @author lijinlong
*
*/
public class SimpleProducer {
/** JNDI name for ConnectionFactory */
static final String CONNECTION_FACTORY_JNDI_NAME = "ConnectionFactory";
/** JNDI name for Queue Destination (use for PTP Mode) */
static final String QUEUE_JNDI_NAME = "exampleQueue";
/** JNDI name for Topic Destination (use for Pub/Sub Mode) */
static final String TOPIC_JNDI_NAME = "exampleTopic";
/**
* 发送消息到队列.<br>
* PTP Mode.
*/
@Test
public void sendToQueue() {
send(QUEUE_JNDI_NAME);
} /**
* 发送消息到主题.<br>
* PTP Mode.
*/
@Test
public void sendToTopic() {
send(TOPIC_JNDI_NAME);
}
/**
* 发送到指定的目的地.
*
* @param destJndiName
* 目的地的JNDI name：{@link #QUEUE_JNDI_NAME}或
* {@link #TOPIC_JNDI_NAME}.
*/
private void send(String destJndiName) {
Context jndiContext = null;
ConnectionFactory connectionFactory = null;
Connection connection = null;
Session session = null;
Destination destination = null;
MessageProducer producer = null;
// create a JNDI API IntialContext object
try {
jndiContext = new InitialContext();
} catch (NamingException e) {
System.out.println("Could not create JNDI Context:"
+ e.getMessage());
System.exit(1);
}
// look up ConnectionFactory and Destination
try {
connectionFactory = (ConnectionFactory) jndiContext
.lookup(CONNECTION_FACTORY_JNDI_NAME);
destination = (Destination) jndiContext.lookup(destJndiName);
} catch (NamingException e) {
System.out.println("JNDI look up failed:" + e.getMessage());
System.exit(1);
}
// send Messages and finally release the resources.
try {
connection = connectionFactory.createConnection();
session = connection.createSession(Boolean.FALSE,
Session.AUTO_ACKNOWLEDGE);
producer = session.createProducer(destination);
TextMessage message = session.createTextMessage();
for (int i = 0; i < 3; i++) {
message.setText(String.format("This is the %dth message.",
i + 1));
producer.send(message);
}
} catch (JMSException e) {
e.printStackTrace();
} finally {
try {
if (session != null)
session.close();
if (connection != null)
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
SimpleProducer.java
3.AutoAckConsumer.java
package cn.sinobest.asj.consumer.jms.acknowledge;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import org.junit.Test;
import cn.sinobest.asj.consumer.util.Hold;
/**
* AUTO_ACKNOWLEDGE确认模式的Consumer.<br>
* 基于PTP Mode，采用异步的方式接收消息，研究抛出或不抛出异常的情况下，Queue中的消息的出队情况.<br>
*
* @author lijinlong
*
*/
public class AutoAckConsumer {
/** JNDI name for ConnectionFactory */
static final String CONNECTION_FACTORY_JNDI_NAME = "ConnectionFactory";
/** JNDI name for Queue Destination (use for PTP Mode) */
static final String QUEUE_JNDI_NAME = "exampleQueue";
/**
* 正常的接收.<br>
*/
@Test
public void receiveNormal() {
MessageListener listener = new MessageListener() {
public void onMessage(Message message) {
try {
String text = ((TextMessage) message).getText();
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
};
receive(listener);
}
/**
* 故意抛出异常的接收.<br>
* 结果：
* <ul>
* <li>JMS Provider重复发送消息给Consumer。重复次数达到一定的阀值，JMS
* Provider认为此消息无法消费，此消息将会被删除或者迁移到"dead letter"通道中。</li>
* <li>在测试过程中，会重发6次（共发7次），然后移到ActiveMQ.DLQ队列；DLQ - dead letter queue.</li>
* <li>重发次数可以配置 -
* 在brokerUrl中指定参数jms.redeliveryPolicy.maximumRedeliveries=3，则重发3次（共4次）.</li>
* </ul>
*/
@Test
public void receiveIntentionalException() {
MessageListener listener = new MessageListener() {
public void onMessage(Message message) {
try {
String text = ((TextMessage) message).getText();
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
boolean intentional = true;
if (intentional) {
throw new RuntimeException("故意抛出的异常。");
}
}
};
receive(listener);
} /**
* 接收消息.<br>
*
* @param listener
* 监听器，如果消息接收成功，将被回调.
*/
private void receive(MessageListener listener) {
Context jndiContext = null;
ConnectionFactory connectionFactory = null;
Connection connection = null;
Session session = null;
Destination destination = null;
MessageConsumer consumer = null;
// create a JNDI API IntialContext object
try {
jndiContext = new InitialContext();
} catch (NamingException e) {
System.out.println("Could not create JNDI Context:"
+ e.getMessage());
System.exit(1);
}
// look up ConnectionFactory and Destination
try {
connectionFactory = (ConnectionFactory) jndiContext
.lookup(CONNECTION_FACTORY_JNDI_NAME);
destination = (Destination) jndiContext.lookup(QUEUE_JNDI_NAME);
} catch (NamingException e) {
System.out.println("JNDI look up failed:" + e.getMessage());
System.exit(1);
}
// receive Messages and finally release the resources.
try {
connection = connectionFactory.createConnection();
connection.start(); // connection should be called in
// receiver-client
session = connection.createSession(Boolean.FALSE,
Session.AUTO_ACKNOWLEDGE);
consumer = session.createConsumer(destination);
// key code for asynchronous receive:set messageListener
consumer.setMessageListener(listener);
Hold.hold(); // 阻塞程序继续执行
} catch (JMSException e) {
e.printStackTrace();
} finally {
try {
if (session != null)
session.close();
if (connection != null)
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
AutoAckConsumer.java
4.ClientAckConsumer.java
package cn.sinobest.asj.consumer.jms.acknowledge;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import org.junit.Test;
import cn.sinobest.asj.consumer.util.Hold;
/**
* CLIENT_ACKNOWLEDGE确认模式的Consumer.<br>
* 基于PTP Mode，采用异步的方式接收消息，研究从不确认、每2次确认的情况下，Queue中的消息的出队情况.<br>
*
* @author lijinlong
*
*/
public class ClientAckConsumer {
/** JNDI name for ConnectionFactory */
static final String CONNECTION_FACTORY_JNDI_NAME = "ConnectionFactory";
/** JNDI name for Queue Destination (use for PTP Mode) */
static final String QUEUE_JNDI_NAME = "exampleQueue";
/**
* 从不确认的接收.<br>
* 结果：
* <ul>
* <li>只接收一次，但是消息不会出队.</li>
* <li>Consumer重启，会再次接收到消息.</li>
* </ul>
*/
@Test
public void receiveWithoutAck() {
MessageListener listener = new MessageListener() {
public void onMessage(Message message) {
try {
String text = ((TextMessage) message).getText();
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
};
receive(listener);
} private int ack_count = 0; // 确认次数统计
/**
* 每接收两次确认一次.<br>
* 结果：每次确认不是只对当前的Message进行确认，而是对自上次确认以来的所有Message进行确认.在这里，每次确认2条.
*/
@Test
public void receivePerTwice() {
MessageListener listener = new MessageListener() {
public void onMessage(Message message) {
try {
String text = ((TextMessage) message).getText();
System.out.println(text); ack_count ++;
if (ack_count % 2 == 0)
message.acknowledge(); } catch (JMSException e) {
e.printStackTrace();
}
}
};
receive(listener);
}
/**
* 接收消息.<br>
*
* @param listener
* 监听器，如果消息接收成功，将被回调.
*/
private void receive(MessageListener listener) {
Context jndiContext = null;
ConnectionFactory connectionFactory = null;
Connection connection = null;
Session session = null;
Destination destination = null;
MessageConsumer consumer = null;
// create a JNDI API IntialContext object
try {
jndiContext = new InitialContext();
} catch (NamingException e) {
System.out.println("Could not create JNDI Context:"
+ e.getMessage());
System.exit(1);
}
// look up ConnectionFactory and Destination
try {
connectionFactory = (ConnectionFactory) jndiContext
.lookup(CONNECTION_FACTORY_JNDI_NAME);
destination = (Destination) jndiContext.lookup(QUEUE_JNDI_NAME);
} catch (NamingException e) {
System.out.println("JNDI look up failed:" + e.getMessage());
System.exit(1);
}
// receive Messages and finally release the resources.
try {
connection = connectionFactory.createConnection();
connection.start(); // connection should be called in
// receiver-client
session = connection.createSession(Boolean.FALSE,
Session.CLIENT_ACKNOWLEDGE);
consumer = session.createConsumer(destination);
// key code for asynchronous receive:set messageListener
consumer.setMessageListener(listener);
Hold.hold(); // 阻塞程序继续执行
} catch (JMSException e) {
e.printStackTrace();
} finally {
try {
if (session != null)
session.close();
if (connection != null)
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
ClientAckConsumer.java
5.DupsOkAckConsumer.java
package cn.sinobest.asj.consumer.jms.acknowledge;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import org.junit.Test;
import cn.sinobest.asj.consumer.util.Hold;
/**
* DUPS_OK_ACKNOWLEDGE确认模式的Consumer.<br>
* @author lijinlong
*
*/
public class DupsOkAckConsumer {
/** JNDI name for ConnectionFactory */
static final String CONNECTION_FACTORY_JNDI_NAME = "ConnectionFactory";
/** JNDI name for Topic Destination (use for Pub/Sub Mode) */
static final String TOPIC_JNDI_NAME = "exampleTopic"; /**
* 从主题接收消息.
*/
@Test
public void receive() {
receive(createMessageListener());
} /**
* 创建MessageListener实例.
* @return
*/
private MessageListener createMessageListener() {
MessageListener listener = new MessageListener() {
public void onMessage(Message message) {
try {
String text = ((TextMessage) message).getText();
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
} try {
Thread.sleep(5 * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}; return listener;
} /**
* 接收消息.<br>
*
* @param listener
* 监听器，如果消息接收成功，将被回调.
*/
private void receive(MessageListener listener) {
Context jndiContext = null;
ConnectionFactory connectionFactory = null;
Connection connection = null;
Session session = null;
Destination destination = null;
MessageConsumer consumer = null;
// create a JNDI API IntialContext object
try {
jndiContext = new InitialContext();
} catch (NamingException e) {
System.out.println("Could not create JNDI Context:"
+ e.getMessage());
System.exit(1);
}
// look up ConnectionFactory and Destination
try {
connectionFactory = (ConnectionFactory) jndiContext
.lookup(CONNECTION_FACTORY_JNDI_NAME);
destination = (Destination) jndiContext.lookup(TOPIC_JNDI_NAME);
} catch (NamingException e) {
System.out.println("JNDI look up failed:" + e.getMessage());
System.exit(1);
}
// receive Messages and finally release the resources.
try {
connection = connectionFactory.createConnection();
connection.start(); // connection should be called in
// receiver-client
session = connection.createSession(Boolean.FALSE,
Session.DUPS_OK_ACKNOWLEDGE);
consumer = session.createConsumer(destination);
// key code for asynchronous receive:set messageListener
consumer.setMessageListener(listener);
Hold.hold(); // 阻塞程序继续执行
} catch (JMSException e) {
e.printStackTrace();
} finally {
try {
if (session != null)
session.close();
if (connection != null)
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
DupsOkAckConsumer.java
AMQ学习笔记 - 16. 确认机制的测试的更多相关文章
-
AMQ学习笔记 - 06. 可靠消息传送
概述 本文介绍JMS中可能发生消息故障的3个隐患阶段,以及确保消息安全的3种保障机制. 故障分析 在介绍可靠传送的确保机制之前,先分析消息在传送的过程中可能在哪个阶段出现问题. 1.两个跃点 跃点的含 ...
-
Storm学习笔记 - 消息容错机制
Storm学习笔记 - 消息容错机制 文章来自「随笔」 http://jsynk.cn/blog/articles/153.html 1. Storm消息容错机制概念 一个提供了可靠的处理机制的spo ...
-
golang学习笔记16 beego orm 数据库操作
golang学习笔记16 beego orm 数据库操作 beego ORM 是一个强大的 Go 语言 ORM 框架.她的灵感主要来自 Django ORM 和 SQLAlchemy. 目前该框架仍处 ...
-
Ext.Net学习笔记16：Ext.Net GridPanel 折叠/展开行
Ext.Net学习笔记16:Ext.Net GridPanel 折叠/展开行 Ext.Net GridPanel的行支持折叠/展开功能,这个功能个人觉得还说很有用处的,尤其是数据中包含图片等内容的时候 ...
-
java学习笔记09--反射机制
java学习笔记09--反射机制 什么是反射: 反射是java语言的一个特性,它允许程序在运行时来进行自我检查并且对内部的成员进行操作.例如它允许一个java的类获取他所有的成员变量和方法并且显示出来 ...
-
SQL反模式学习笔记16 使用随机数排序
目标:随机排序,使用高效的SQL语句查询获取随机数据样本. 反模式:使用RAND()随机函数 SELECT * FROM Employees AS e ORDER BY RAND() Limit 1 ...
-
AMQ学习笔记 - 17. 事务的测试
概述 对事务机制进行测试. 测试实例 测试实例 结果预测 发送正常 3条消息入队 发送异常 0条消息入队 接收正常 3条消息出队 接收异常 0条消息出队 demo设计 设计图 测试分工 测试类 测试方 ...
-
AMQ学习笔记 - 21. 异步发送
原文地址:Async Sends 背景 ActiveMQ支持同步.异步两种发送的模式将消息发送到broker,模式的选择对发送延时有巨大的影响.producer能达到怎样的产出率[1],主要受发送延时 ...
-
java学习笔记13--反射机制与动态代理
本文地址:http://www.cnblogs.com/archimedes/p/java-study-note13.html,转载请注明源地址. Java的反射机制 在Java运行时环境中,对于任意 ...
随机推荐
-
Windows Server 2012安装时所需要的KEY
Windows Server 2012不像Server 2008和2008 R2那样可以先装系统再输入序列号,而是在一开始就必须输入Server 2012 cdkey,目前在网上找到两枚序列号,标准版 ...
-
SQL 调试：无法启动 T-SQL 调试。未能附加到 SQL Server 进程
将 Windows 登录帐户添加为 sysadmin 已经具有 sysadmin 特权的用户必须执行以下命令: sp_addsrvrolemember 'Domain\Name', 'sysadmin ...
-
PHP无限极分类生成树方法
你还在用浪费时间又浪费内存的递归遍历无限极分类吗,看了该篇文章,我觉得你应该换换了.这是我在OSChina上看到的一段非常精简的PHP无限极分类生成树方法,整理分享了. function genera ...
-
Java eclipse生成doc文档
这里讲解下eclipse成为doc文档,首先代码: /** * @author szy * @version 1.0 */ package com.founder.sun; class Cat{ pu ...
-
一个能拖动，能调整大小，能更新bind值的vue指令-vuedragx
一. 背景说明 开发一个可自定义组件化门户配置页面,期间采用了vue框架作为前端视图引擎,作为一个刚入手vue的萌新,开发第一个功能就遇到了拦路虎.需要一个拖动并且可改变大小的容器盒子.当时查看vue ...
-
修改任务显示WrkTaskIp.aspx页面
环境:Sharepoint2010 需求:在审批任务页面中插入或显示表单或列表内容,让内容与审核在同一页面上. 修改文件:WrkTaskIp.aspx(改前记得备份) 文件所在的路径:C:\Progr ...
-
MDC的使用（Mapped Diagnostic Context）的使用
通常我们可能会有大量的任务需要提交提交到线程池执行,但是此时如果不对日志添加唯一标识进行区分的话回到错乱一坨无法进行查看.因此可以对每一天日志添加唯一的标识,例如使用userid作为日志的唯一标志.这 ...
-
eclipse jsp:useBean搞死人了。
eclipse jsp:useBean搞死人了 首页,用eclipse需要经常重启tomcat服务器,这是因为你编辑了页面正浏览这个页面,而这个页面还处在之前的错误编译中... 其次,第一次用到use ...
-
C结构体
什么是结构体? 简单的来说,结构体就是个能够包含不同数据类型的一个结构,他是一种能够自己定义的数据类型,他的特点和数组主要有两点不同,首先结构体能够在一个结构中声明不同的数据类型,第二相同结构的结构体 ...
-
[置顶] CSS语言精粹
本文主要是对CSS中一些比较重要的高级部分作了一些整理,这些内容也许不是经常使用,但是都很强大.本文将长期更新. 边框 CSS2.1 规定:元素的背景是内容.内边距和边框区的背景.设置背景颜色时是没有 ...