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

FunTester · 2020年03月31日 · 1488 次阅读

本人在负责服务压测的实践中,遇到了一个需求,就是对消息队列的 dubbo 接口性能进行压测,主要分两类:一类是往队列里面添加,一类是从队列中取值(等同删除)。是一个 server 的两个不同方法。同组其他人用的 jmeter 进行的 dubbo 接口压测。
队列的添加规则比较简单,主要有一个标志 msg,由事件类型 + 用户标识符 + 消息体构成。做此类此类测试的时候遇到的问题就是如果构建消息体,每次都构建不同的消息体,这里我才用了纳秒 + 随机数的方式,后来发现直接用纳秒就行。(这里相信 jmeter 也应该有响应的方法)

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


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() 方法里面为每一个线程进行数据初始化操作。

  • 郑重声明:文章首发于公众号 “FunTester”,欢迎关注交流,禁止第三方(腾讯云除外)转载、发表。

技术类文章精选

无代码文章精选

如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
暂无回复。
需要 登录 后方可回复, 如果你还没有账号请点击这里 注册