어떻게 메시지 큐 에 대해 성능 테스트 를 합 니까?

본인 은 서비스 압력 측정 을 담당 하 는 실천 에서 하나의 수 요 를 만 났 습 니 다. 바로 메시지 대기 열의 dubbo 인터페이스 성능 을 압력 측정 하 는 것 입 니 다. 주로 두 가지 로 나 뉘 는데 하 나 는 대기 열 에 추가 하 는 것 이 고 하 나 는 대기 열 에서 값 을 추출 하 는 것 입 니 다 (같은 삭제).서버 의 두 가지 다른 방법 입 니 다.같은 그룹의 다른 사람들 이 사용 하 는 jmeter 에서 진행 하 는 dubbo 인터페이스 압력 측정.대기 열의 추가 규칙 은 비교적 간단 합 니 다. 주로 하나의 표지 msg 가 있 는데 이벤트 유형 + 사용자 식별 자 + 메시지 체 로 구성 되 어 있 습 니 다.이런 테스트 를 할 때 발생 하 는 문 제 는 메시지 체 를 구축 하면 매번 다른 메시지 체 를 구축 하 는 것 이다. 여기 서 나 는 나 초 + 랜 덤 수의 방식 을 사 용 했 는데 나중에 나 초 를 직접 사용 하면 된다 는 것 을 알 게 되 었 다.(jmeter 도 응답 하 는 방법 이 있 을 거 라 고 믿 습 니 다)
대기 열 을 추가 하 는 테스트 에서 jmeter 가 어떻게 실현 되 는 지 잘 모 르 겠 습 니 다. 그들 이 직접 포기 하기 때문에 제 가 사용 하 는 방안 은 충분 한 양의 메 시 지 를 구축 한 다음 에 메시지 데 이 터 를 꺼 내 서 스 레 드 안전 한 집합 에 넣 고 다 중 스 레 드 를 가 져 오 는 것 입 니 다. 자바 의 링크 드 BlockingQueue 메시지 대기 열 을 사용 합 니 다.테스트 를 한 번 도 끝내 지 못 하고 테스트 데 이 터 를 리 셋 하여 중간 에 실패 하 는 것 을 방지 합 니 다.

public int createQ() {
        String absolutePath = new File("").getAbsolutePath();
        List 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 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 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 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 msgs = addmsg();


    public static LinkedBlockingQueue addmsg() {
        String absolutePath = new File("").getAbsolutePath();
        List strings = WriteRead.readTxtFileByLine(absolutePath + "/queue");
        LinkedBlockingQueue ss = new LinkedBlockingQueue<>();
        ss.addAll(strings);
        logger.info("       ");
        return ss;
    }

4. 567917. 여기 서 문제 가 있 습 니 다. 계속 테스트 하 는 과정 에서 addmsg 방법 은 테스트 과정 에서 실 행 될 수 있 습 니 다
제 가 테스트 를 할 때 데이터 양 이 충분 하기 때문에 처 리 를 하지 않 았 습 니 다. 만약 에 데이터 양 이 여러 번 테스트 를 지탱 하지 못 하면 테스트 를 시작 하기 전에 msgs 를 초기 화하 거나 before () 방법 에서 모든 스 레 드 에 데 이 터 를 초기 화 할 수 있 습 니 다.
관심 있 는 동 화 를 환영 합 니 다.