如何用python做简单的接口压力测试
作者：星旅zl
最近研究了一下接口的压力测试，主要来说就是连续频繁的对接口的调用，来测试接口的响应速度、返回结果，找到接口的性能瓶颈，最大承受极限等。
做接口压力测试的方法和工具很多，比较常用的工具有postman、jmeter,这两种都是界面形式的操作
当然也可以使用代码来跑~~我怀疑你在装逼，但是没有证据！
那接下来就是如何用python脚本进行接口测试了，这里使用的是python3,相关的下载安装和环境配置这里不做多说，很多文章都有介绍。
首先我们先要了解要用到http.client.HTTPConnection网络请求，可以去这篇文章看具体使用方法
然后就是需要到线程的支持，最后是对执行结果的输出分析。
注：下文分段解释，但是执行时请放在一个.py文件下（代码有参考借鉴一些别的文章）
一、引入所需库和设置需要到的参数
import threading
import time
import http.client
import urllib
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
pararms = urllib.parse.urlencode({'token':'xxx'})
rqheaders={ #'DeviceToken':'xxxxxxxxx','OSVersion':'1.0.3','AppVersion':'14',
'Accept':'application/json, text/plain, */*',
'Accept-Encoding':'gzip, deflate, br',
'Accept-Language':'zh-CN,zh;q=0.9',
'Host':'deve.xxx.com',
'Origin':'https://www.xxx.com',
'Referer':'https://www.xxx.com/xxx_test/',
'token':'xxx',
'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.157 Safari/537.36'}
HOST1 = "www.baidu.com"; #主机地址 例如192.168.1.101 nanning.anjuke.com,
HOST_L = "deve.xxx.com"
HOST_IP = "xxx.xx.xxx.xx" #ip地址
PORT = 9999 #端口
METHOD = {1:"GET",2:"POST",3:"OPTIONS"}
URI = "/api/web/XXX" #相对地址
TOTAL = 0 #总数
SUCC = 0 #响应成功数
FAIL = 0 #响应失败数
EXCEPT = 0 #响应异常数
MAXTIME=0 #最大响应时间
MINTIME=100 #最小响应时间，初始值为100秒
GT3=0 #统计3秒内响应的
LT3=0 #统计大于3秒响应的
# 并发的线程数
thread_count = 5这里请根据真实的接口地址进行配置，有需要到token的请自行拿到token值，不需要头部的的可以在下面中写headers = {}
二、创建一个 threading.Thread 的派生类
class RequestThread(threading.Thread):
# 构造函数
def __init__(self, thread_name):
threading.Thread.__init__(self)
self.test_count = 0
print ("===========task init===========METHOD=",METHOD[3])
# 线程运行的入口函数
def run(self):
self.test_performace()
def test_performace(self):
global TOTAL
global SUCC
global FAIL
global EXCEPT
global GT3
global LT3
try:
st = time.time()
conn = http.client.HTTPConnection(HOST1,PORT) #若没有PORT请删掉这个参数
req = conn.request(METHOD[1], URI, body = {}, headers = rqheaders) # or POST headers=rqheaders
res = conn.getresponse()
#print ('msg:', res.msg)
#print ('headers:', res.getheaders())
print("res.status =",res.status)
print('res.read =',res.read().decode('utf-8'))
start_time
if res.status == 200:
TOTAL = TOTAL+1
SUCC = SUCC+1
print("TOTAL = ",TOTAL)
else:
TOTAL = TOTAL+1
FAIL = FAIL+1
print("TOTAL = ",TOTAL)
time_span = time.time()-st
print ("%s:%f\n"%(self.name,time_span))
self.maxtime(time_span)
self.mintime(time_span)
if time_span>3:
GT3 = GT3+1
else:
LT3 = LT3+1
except Exception as e:
print("e =",e)
TOTAL = TOTAL+1
EXCEPT = EXCEPT+1
conn.close()
def maxtime(self,ts):
global MAXTIME
print(ts)
if ts>MAXTIME:
MAXTIME=ts
def mintime(self,ts):
global MINTIME
if ts<MINTIME:
MINTIME=tsres.read().decode('utf-8')这个方法会返回接口响应的所有内容
三、使用派生类方法写执行
# main 代码开始
print ("===========task start===========")
# 开始的时间
start_time = time.time()
i = 1
while i <= thread_count:
t = RequestThread("thread" + str(i))
t.start()
i = i + 1
t=0
#并发数所有都完成或大于60秒就结束
while TOTAL<thread_count|t>60:
print ("total:%d,succ:%d,fail:%d,except:%d\n"%(TOTAL,SUCC,FAIL,EXCEPT))
print (HOST,URI)
t = t+1
time.sleep(2)
time.sleep(2)
print ("===========task end===========")
print ("total:%d,succ:%d,fail:%d,except:%d"%(TOTAL,SUCC,FAIL,EXCEPT))
print ("response maxtime:",MAXTIME)
print ("response mintime:",MINTIME)
print ("great than 3 seconds:%d,percent:%0.2f"%(GT3,float(GT3)/TOTAL))
print ("less than 3 seconds:%d,percent:%0.2f"%(LT3,float(LT3)/TOTAL))四、上面就是所有的代码
然后我们打开cmd来运行这个脚本，这里设置thread_count = 1，即执行一次请求
cmd进入到相应的py文件夹下，运行脚本结果如下所示：
F:\pythonTest>python testInterfacePressure.py
===========task start===========
===========task init===========METHOD= OPTIONS
res.status = 200
res.read = {..........，"code":1,"message":"数据读取成功","status":200}TOTAL = 1
Thread-1:0.1499270.14992666244506836
===========task end===========
total:1,succ:1,fail:0,except:0
response maxtime: 0.14992666244506836
response mintime: 0.14992666244506836
great than 3 seconds:0,percent:0.00
less than 3 seconds:1,percent:1.00
总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。