Performance Testing of Dubbo Queue Interfaces Using JMeter and Custom Java Concurrency

The author needed to conduct load testing for two Dubbo service methods that interact with a message queue: one for adding messages and one for deleting (or consuming) them. While teammates used JMeter for Dubbo interface testing, the author implemented a custom solution.

For the add operation, a simple message format is used, consisting of an event type, a user identifier, and the message body. Initially the author combined System.nanoTime() with a random number to ensure uniqueness, but later found that the nanosecond timestamp alone was sufficient. The test data is prepared in advance: a large number of messages are generated, placed into a thread‑safe collection, and multiple threads retrieve messages from a LinkedBlockingQueue<String> during the test. After each test run the data is reset to avoid interference from previous failures.

Below is the Java implementation for creating a queue entry:

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;
}

The delete operation polls a message from the shared queue and sends a delete request:

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;
}

The shared message queue is initialized as a static LinkedBlockingQueue<String> populated by the addmsg() method, which reads lines from a file and loads them into the queue:

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;
}

A potential issue identified is that the addmsg() method might be invoked during an ongoing test, leading to unexpected behavior. The author suggests initializing the msgs queue before the test starts or performing per‑thread initialization in the before() method.

Because the author used a sufficiently large data set, no special handling was required for the presented tests; however, for smaller data volumes, the recommended initialization steps become necessary.

Readers are invited to view the original article and discuss further.