Python并行调用本地API打满10核CPU并统计吞吐量的实现咨询

可直接运行的实现代码 #

import asyncio, time
from api import API, DocumentParameters, EndException

# 可调整参数
INIT_CONCURRENT = 2
INCREASE_STEP = 5
INCREASE_INTERVAL = 10  # 单位：秒
MAX_CONCURRENT = 100

# 全局统计变量
success_count = 0
current_concurrent = INIT_CONCURRENT

def call_word_api(data):
    api = API(url='http://localhost:8181/rest/words')
    params = DocumentParameters()
    params["content"] = data
    return api.words(params)

def call_lang_api(data):
    api = API(url='http://localhost:8181/rest/languages')
    params = DocumentParameters()
    params["content"] = data
    return api.language(params)

async def request_task(sem, data, api_func):
    global success_count
    async with sem:
        try:
            await asyncio.to_thread(api_func, data)
            success_count += 1
        except EndException:
            pass

async def monitor():
    global current_concurrent
    last_count = 0
    last_increase_time = time.time()
    while True:
        await asyncio.sleep(1)
        qps = success_count - last_count
        last_count = success_count
        print(f"并发数: {current_concurrent} | 每秒吞吐量(QPS): {qps} | 累计请求数: {success_count}")
        # 定时提升并发上限
        if time.time() - last_increase_time > INCREASE_INTERVAL and current_concurrent < MAX_CONCURRENT:
            current_concurrent += INCREASE_STEP
            last_increase_time = time.time()
            print(f"=== 提升并发上限到 {current_concurrent} ===")

async def main():
    with open("/Users/me/stuff.txt", "r") as f:
        data = f.read()
    asyncio.create_task(monitor())
    # 无限发起请求
    while True:
        sem = asyncio.Semaphore(current_concurrent)
        await asyncio.gather(
            request_task(sem, data, call_word_api),
            request_task(sem, data, call_lang_api),
            return_exceptions=True
        )

if __name__ == "__main__":
    asyncio.run(main())

问题解决说明 #

1、CPU打满实现逻辑

• 初始并发数设为2，每10秒自动增加5个并发上限，你可以根据实际运行情况调整INCREASE_STEP和INCREASE_INTERVAL参数，加快或减慢并发提升速度
• 用信号量控制同时发起的请求数量，并发数上升后本地Java服务会占用更多CPU资源，直到10核全部满载，MAX_CONCURRENT参数用来防止请求过多把Java服务打挂，可以根据实际情况调整
• 如果需要Python端也占用CPU，可以把asyncio.to_thread替换为asyncio.get_event_loop().run_in_executor指定多进程池即可

2、吞吐量统计实现逻辑

• 全局计数器记录所有成功完成的请求数
• 监控协程每秒统计一次过去1秒的成功请求增量，即为当前吞吐量(QPS)
• 控制台会持续打印当前并发数、QPS、累计请求数，你可以对照htop的CPU占用数据，直接记录不同并发度下CPU占用和吞吐量的对应关系

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