How to Implement Fixed‑QPS Load Testing and Fix Common Bugs

Fixed QPS Practice

The author revisits a previously published "Fixed QPS Load‑Testing Model" and presents a concrete implementation for HTTP request load testing using a list of two requests and a fixed‑QPS concurrency controller.

def requests = []
def base = getBase()
def order = new Order(base)
order.create(21, 17, "FunTester", "", 1, 1)
requests << FanLibrary.lastRequest
order.edit()
requests << FanLibrary.lastRequest

def threads = []

requests.size().times {
    threads << new RequestTimesFixedQps<>(5, 50, new HeaderMark("requestid"), requests[it])
}

new FixedQpsConcurrent(threads, "Fixed QPS Practice").start()

allOver()

A generic internal‑class script is also shared; only the doing() and clone() methods need to be implemented.

public static void main(String[] args) {
    TT qps = new TT(5, 50);
    new FixedQpsConcurrent(qps).start();
}

static class TT extends FixedQpsThread {
    public TT(int i, int i1) {
        super(null, i1, i, null, true);
    }

    @Override
    protected void doing() throws Exception {
        sleep(1);
    }

    @Override
    public TT clone() {
        return this;
    }
}

Async Compensation Thread Pool State Validation

The original compensation thread code could throw a RejectedExecutionException when the thread pool had already been shut down.

@Override
public void run() {
    logger.info("补偿线程开始!");
    while (key) {
        sleep(HttpClientConstant.LOOP_INTERVAL);
        int actual = executeTimes.get();
        int qps = baseThread.qps;
        long expect = (Time.getTimeStamp() - FixedQpsConcurrent.this.startTime) / 1000 * qps;
        if (expect > actual + qps) {
            logger.info("期望执行数:{},实际执行数:{},设置QPS:{}", expect, actual, qps);
            range((int) expect - actual).forEach(x -> {
                sleep(100);
                executorService.execute(threads.get(this.i++ % queueLength).clone());
            });
        }
    }
    logger.info("补偿线程结束!");
}

The fix adds a check for executorService.isShutdown() before submitting new tasks, preventing the exception.

@Override
public void run() {
    logger.info("补偿线程开始!");
    try {
        while (key) {
            sleep(HttpClientConstant.LOOP_INTERVAL);
            int actual = executeTimes.get();
            int qps = baseThread.qps;
            long expect = (Time.getTimeStamp() - FixedQpsConcurrent.this.startTime) / 1000 * qps;
            if (expect > actual + qps) {
                logger.info("期望执行数:{},实际执行数:{},设置QPS:{}", expect, actual, qps);
                range((int) expect - actual).forEach(x -> {
                    sleep(100);
                    if (!executorService.isShutdown())
                        executorService.execute(threads.get(this.i++ % queueLength).clone());
                });
            }
        }
        logger.info("补偿线程结束!");
    } catch (Exception e) {
        logger.error("补偿线程发生错误!", e);
    }
}

The added executorService.isShutdown() verification ensures tasks are only submitted when the pool is active.

Fix Counter Bug

The original implementation incremented the global execution counter after the doing() method, which could cause the asynchronous compensation thread to over‑compensate.

@Override
public void run() {
    try {
        before();
        threadmark = this.mark == null ? EMPTY : this.mark.mark(this);
        long s = Time.getTimeStamp();
        doing();
        long e = Time.getTimeStamp();
        long diff = e - s;
        FixedQpsConcurrent.executeTimes.getAndIncrement();
        FixedQpsConcurrent.allTimes.add(diff);
        if (diff > HttpClientConstant.MAX_ACCEPT_TIME)
            FixedQpsConcurrent.marks.add(diff + CONNECTOR + threadmark);
    } catch (Exception e) {
        FixedQpsConcurrent.errorTimes.getAndIncrement();
        logger.warn("执行任务失败！,标记:{}", threadmark, e);
    } finally {
        after();
    }
}

The revised code moves the counter increment to occur before the doing() call, eliminating the timing discrepancy.

@Override
public void run() {
    try {
        before();
        threadmark = this.mark == null ? EMPTY : this.mark.mark(this);
        FixedQpsConcurrent.executeTimes.getAndIncrement();
        long s = Time.getTimeStamp();
        doing();
        long e = Time.getTimeStamp();
        long diff = e - s;
        FixedQpsConcurrent.allTimes.add(diff);
        if (diff > HttpClientConstant.MAX_ACCEPT_TIME)
            FixedQpsConcurrent.marks.add(diff + CONNECTOR + threadmark);
    } catch (Exception e) {
        FixedQpsConcurrent.errorTimes.getAndIncrement();
        logger.warn("执行任务失败！,标记:{}", threadmark, e);
    } finally {
        after();
    }
}