Python实现接口并发压力测试脚本详解
发布时间：2026-10-13 |
4人浏览过
本文介绍利用Python进行接口并发压力测试的方法，涵盖并发数、响应时间、吞吐量及性能计数器等核心指标。通过示例脚本演示如何构建多线程压测工具，计算QPS与错误率，并解析Linux系统资源监控命令，帮助开发者掌握接口性能评估的关键技术。
接口性能测试核心指标解析
评估网站性能时，通常关注几个关键指标：并发处理能力、响应速度、系统吞吐量以及各类性能计数器数据。
1、并发数
并发数代表系统在单位时间内能够同时处理的请求总量，该数值直接反映了系统的负载承载能力。
2、响应时间
响应时间是衡量系统性能的核心指标之一，其数值大小直观体现了系统的快慢。具体而言，它指从发起一个请求开始，到最终接收到完整响应数据所经历的总时长。
3、吞吐量
吞吐量表示系统在单位时间内处理请求的数量，是体现系统处理能力的常用性能测试指标。常用的量化指标包括QPS（每秒查询数）、TPS（每秒事务数）以及HPS（每秒HTTP请求数）。吞吐量与并发数、响应时间密切相关，三者之间的数学关系为：QPS（或TPS）等于并发数除以平均响应时间。
4、性能计数器
性能计数器用于描述服务器或操作系统的各项性能数据，例如内存使用量、进程运行时间等。在性能测试中，它们起到“监控与分析”的作用，特别是在分析系统可扩展性及定位性能瓶颈时至关重要。在Linux系统中，可以通过top或uptime命令查看当前的系统负载及资源利用率。资源利用率是指系统各类资源的使用情况，例如CPU占用率为68%，内存占用率为55%，其计算方式通常为“资源实际使用量除以总的可用资源量”。
Python并发压测脚本实现
以下是用于下单接口的压测脚本示例：
#!/usr/bin/env python
#-*- coding:utf-8 -*-
import requests,time,json,threading,random
class Presstest(object):
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.121 Safari/537.36',
'Content-Type': 'application/json; charset=UTF-8',
}
def __init__(self,login_url,press_url,phone="1376193000",password="123456"):
self.login_url = login_url
self.press_url = press_url
self.phone = phone
self.password = password
self.session = requests.Session()
self.session.headers = self.headers
def login(self):
'''登陆获取session'''
data = data = {'t': int(time.time() * 1000), 'userName': self.phone, 'passWord': self.password}
res = self.session.post(self.login_url,data=json.dumps(data))
XToken = res.json().get('data').get('companyToken')
self.session.headers['X-Token'] = XToken
def testinterface(self):
'''压测接口'''
self.session.headers['X-UnionId'] = 'of6uw1CUVhP533sQok'
data = {"id": int(''.join(str(random.choice(range(10))) for _ in range(10))),
"openId": "oMr0c5LGJjlTc", "addressId": 10, "shipType": "SELF", "totalAmount": 5,
"receivable": 5, "carts": [
{"amount": 1, "barcode": "1234567890", "skuId": 1, "spec": "34", "itemAmount": 5, "price": 0,
"cover": "xxxx-dd.oss-cn-shanghai.aliyuncs.com/dfc91fd067ac464c096c90af33a196a5.png",
"name": "沙宣洗发水", "packingType": "瓶", "placeOfOrigin": "上海", "productId": "310153323435134976",
"retailPrice": 5, "suitableAge": "1-100"}], "formId": "the formId is a mock one", "comments": "aa"}
global ERROR_NUM
try:
html = self.session.post(self.press_url, data=json.dumps(data))
if html.json().get('code') != 0:
print(html.json())
ERROR_NUM += 1
except Exception as e:
print(e)
ERROR_NUM += 1
def testonework(self):
'''一次并发处理单个任务'''
i = 0
while i < ONE_WORKER_NUM:
i += 1
self.work()
time.sleep(LOOP_SLEEP)
def run(self):
'''使用多线程进程并发测试'''
t1 = time.time()
Threads = []
for i in range(THREAD_NUM):
t = threading.Thread(target=self.testonework, name="T" + str(i))
t.setDaemon(True)
Threads.append(t)
for t in Threads:
t.start()
for t in Threads:
t.join()
t2 = time.time()
print("===============压测结果===================")
print("URL:", self.press_url)
print("任务数量:", THREAD_NUM, "*", ONE_WORKER_NUM, "=", THREAD_NUM * ONE_WORKER_NUM)
print("总耗时(秒):", t2 - t1)
print("每次请求耗时(秒):", (t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM))
print("每秒承载请求数:", 1 / ((t2 - t1) / (THREAD_NUM * ONE_WORKER_NUM)))
print("错误数量:", ERROR_NUM)
if __name__ == '__main__':
login_url = 'https://ds.xxxxx.com/sys/sysUser/login'
press_url = 'https://ds.xxxxx.com/weshop/order/checkout'
phone = "1376193000"
password = "123456"
THREAD_NUM = 1 # 并发线程总数
ONE_WORKER_NUM = 5 # 每个线程的循环次数
LOOP_SLEEP = 0.1 # 每次请求时间间隔(秒)
ERROR_NUM = 0 # 出错数
obj = Presstest(login_url=login_url,press_url=press_url,phone=phone,password=password)
obj.login()
obj.run()压测结果数据分析
执行上述脚本后的输出结果如下：
===============压测结果===================
URL: https://ds.xxxxx.com/weshop/order/checkout
任务数量: 1 * 5 = 5
总耗时(秒): 1.9810078144073486
每次请求耗时(秒): 0.39620156288146974
每秒承载请求数: 2.5239678327547805
错误数量: 0以上内容为本文的全部介绍，希望能为读者的学习提供有益帮助，同时也欢迎大家继续支持本站。
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
