代码片段利用python多进程进行接口并发测试

利用 python 多进程进行接口并发测试

前言

最近公司服务类的项目有个测试并发的需求，测试的目的并非是服务器能抗住多大的并发，而是测试服务在并发请求下是否是阻塞的。比如单个并发的执行时间大搞是两秒的话，那么 10 个并发的总执行时间是所有请求执行时间的总和还是最大请求时间。

代码

import time
import requests
from multiprocessing import Process
from multiprocessing import Pool

data = {
    "times": 10, # 并发量
    "method": "POST",
    "url": "http://xxx.com/xxx",
    "header": {
        "Content-Type": "application/json",
        "user-agent": "python-mock/0.0.1"
    },
    "body": {# 参数}
}

def run_task(idx):
    response = requests.post(data["url"], json=data["body"], headers=data["header"])
    if response.status_code == 200:
        result = response.content.decode('utf-8')
    else:
        result = "访问失败"
    print("第 %s 次执行：%s \n" % (idx, result))

if __name__ == '__main__':
    p = Pool(data["times"])
    for index in range(data["times"]):
        p.apply_async(run_task, args=(index + 1,))

    p.close()
    p.join()
    print("执行结束.")

实现原理是利用 python 的多进程，测试 10 个并发量就利用 for 循环创建 10 个进程。
p.join() 是阻塞等待所有进程执行结束。