如何按预设请求数量自动终止Locust接口压测进程

• 随机任务调度逻辑问题：你给两个接口方法都加了@task装饰器，Locust默认会随机调度两个任务执行，无法保证两个API都各跑完10次
• 循环逻辑错误：range(10)生成的是0-9的序列，你在i==9时直接调用self.interrupt()跳过了第10次请求，实际每个API只跑了9次
• 终止逻辑不符合预期：self.interrupt()只会终止当前TaskSet的执行，不会触发整个Locust进程退出，进程会持续运行等待新任务
• 没有全局完成校验：缺少全局计数逻辑判断所有API的所有请求是否都已完成，无法触发自动终止

import json
import random
from threading import Lock
from locust import HttpUser, TaskSet, between, events

# 配置项：根据实际API数量调整
API_COUNT = 2
PER_API_RUN_TIMES = 10
# 总请求数，多并发场景下需要乘以并发用户数
TOTAL_REQUESTS = API_COUNT * PER_API_RUN_TIMES

# 全局计数器&锁，避免多并发计数错误
completed_requests = 0
counter_lock = Lock()

# 监听请求完成事件，统计执行进度
@events.request.add_listener
def on_request_complete(request_type, name, response_time, response_length, response, context, exception, **kwargs):
    global completed_requests
    with counter_lock:
        # 不需要区分请求成功/失败就删除下方if判断
        if not exception:
            completed_requests += 1
        # 所有请求执行完成，终止整个Locust进程
        if completed_requests >= TOTAL_REQUESTS:
            context.environment.runner.quit()

class UserBehavior(TaskSet):
    def on_start(self):
        # 初始化公共参数
        self.camp_code = random.choice(t.camp_code)
        self.pgm_id = random.choice(t.pgm_id)
        # 按顺序执行所有API的请求，保证每个API都能跑满次数
        self.alloction_details()
        self.get_content_details()
        self.interrupt()

    def alloction_details(self):
        pgm_id = self.pgm_id
        camp_code = self.camp_code
        print("L21", camp_code)
        myheaders = {'Content-Type': 'application/json', 'Accept': 'application/json'}
        load = {
            "Code": str(camp_code),
            "Id": str(pgm_id)
        }
        # 执行10次当前接口请求
        for _ in range(PER_API_RUN_TIMES):
            with self.client.post('/ContentDetails', data=json.dumps(load), headers=myheaders, catch_response=True) as response:
                print(response.status_code)
                if response.status_code == 200:
                    response.success()
                    print("Response of allocation details ", response.text, '\n\n')
                else:
                    response.failure('API failure')

    def get_content_details(self):
        myheaders = {'Content-Type': 'application/json', 'Accept': 'application/json'}
        pgm_id = self.pgm_id
        camp_code = self.camp_code
        # 执行10次当前接口请求
        for _ in range(PER_API_RUN_TIMES):
            with self.client.post(f'/ScheduleDetails?Code={str(camp_code)}', catch_response=True) as response:
                print(response.status_code)
                if response.status_code == 200:
                    response.success()
                    print("Response : ", response.text, '\n\n')
                else:
                    response.failure('API failure')

class WebsiteUser(HttpUser):
    tasks = [UserBehavior]
    host = 'localhost'
    wait_time = between(1, 3)

• 单用户场景下直接执行命令 locust -f 你的脚本文件名.py --headless -u 1 -r 1 即可自动运行
• 多并发场景下需要将TOTAL_REQUESTS调整为 API_COUNT * PER_API_RUN_TIMES * 并发用户数
• 所有请求执行完成后进程会自动退出，无需手动中断

内容的提问来源于stack exchange，提问作者phani