vlambda博客
学习文章列表

如何对消息队列做性能测试

本人在负责服务压测的实践中,遇到了一个需求,就是对消息队列的dubbo接口性能进行压测,主要分两类:一类是往队列里面添加,一类是从队列中取值(等同删除)。是一个server的两个不同方法。同组其他人用的jmeter进行的dubbo接口压测。

队列的添加规则比较简单,主要有一个标志msg,由事件类型+用户标识符+消息体构成。做此类此类测试的时候遇到的问题就是如果构建消息体,每次都构建不同的消息体,这里我才用了纳秒+随机数的方式,后来发现直接用纳秒就行。(这里相信jmeter也应该有响应的方法)


在添加队列的测试不太清楚jmeter如何实现,因为他们直接放弃掉了,我才用的方案是,先构建足够多数量的消息,然后将消息数据拿出来放到一个线程安全的集合中,多线程去拿,使用的是java的LinkedBlockingQueue<String>消息队列。没做完一次测试,重置一次测试数据,防止中途有失败的情况。


public int createQ() { String absolutePath = new File("").getAbsolutePath(); List<String> strings = WriteRead.readTxtFileByLine(absolutePath + "/dubbo"); new Concurrent(new ThreadBase(SourceCode.changeStringToInt(strings.get(0))) { @Override protected void before() {
}
@Override protected void doing() throws Exception { CreateQueueRequest createQueueRequest = new CreateQueueRequest(); createQueueRequest.setReqId(TraceKeyHolder.getTraceKey()); createQueueRequest.setDelayTime(System.currentTimeMillis() + 3600 * 1000); String msg = "wait_for_publish:8888" + "@" + System.nanoTime() + PublishType.ZUOYE; createQueueRequest.setMsg(msg); createQueueRequest.setTaskTypeEnum(TaskTypeEnum.PUBLISH_PROMU); createQueueRequest.setTtl(0L); CommonResponse<CreateQueueResultVo> queue = commonDelayQueueService.createQueue(createQueueRequest); logger.info("createQueue0 {}", JsonUtil.obj2Json(queue)); }
@Override protected void after() {
} }, SourceCode.changeStringToInt(strings.get(1))).start(); return 0;    }

删除队列:

public int deleteQ() throws InterruptedException { if (msgs.size() == 0) { logger.info("队列为空了"); msgs = addmsg(); } String absolutePath = new File("").getAbsolutePath(); List<String> strings = WriteRead.readTxtFileByLine(absolutePath + "/dubbo");
new Concurrent(new ThreadBase(SourceCode.changeStringToInt(strings.get(0))) { @Override protected void before() {
}
@Override protected void doing() throws Exception { String msg = msgs.poll(100, TimeUnit.MILLISECONDS); logger.info("msg:{}", msg); DeleteQueueRequest deleteQueueRequest0 = new DeleteQueueRequest(); deleteQueueRequest0.setMsg(msg); deleteQueueRequest0.setTaskTypeEnum(TaskTypeEnum.PUBLISH_PROMU); CommonResponse<String> queue3 = commonDelayQueueService.deleteQueue(deleteQueueRequest0); logger.info("deleteQueue2 {}", JsonUtil.obj2Json(queue3)); }
@Override protected void after() {
} }, SourceCode.changeStringToInt(strings.get(1))).start();
return 0; }

其中msgs的设置如下:

public static LinkedBlockingQueue<String> msgs = addmsg();

public static LinkedBlockingQueue<String> addmsg() { String absolutePath = new File("").getAbsolutePath(); List<String> strings = WriteRead.readTxtFileByLine(absolutePath + "/queue"); LinkedBlockingQueue<String> ss = new LinkedBlockingQueue<>(); ss.addAll(strings); logger.info("重新读取队列值"); return ss; }
  •  这里会有一个问题:在不断测试过程中,addmsg方法可能在测试过程中被执行


因为我在做测试的时候,数据量足够大,所以没有做处理,如果数据量不足以支撑很多次测试,可以采用启动测试前把msgs进行初始化,或者在before()方法里面为每一个线程进行数据初始化操作。