一个性能测试工具aiohttp实现
#! /usr/bin/python3
# -*- coding:utf-8 -*-
# file: test.py
# author: wangchenxi
# mail: wongchenxi@icloud.com
# brief:
# version: 0.1.00
# Create Time:2021-01-15 21:01:12
# Last Update: 2021-01-16 22时06分19秒
import time
import asyncio
from threading import Thread
import functools
import os
import signal
import aiohttp
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/76.0.3809.100 Safari/537.36'
}
stop_sign = False
def exit_loop(sign_name):
global stop_sign
if not stop_sign:
stop_sign = True
async def task_stop():
loop = asyncio.get_event_loop()
loop.stop()
async def task(e, i, m, u, h, data, loopargs):
url = u
try:
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=0.3)) as s:
if hasattr(s, m):
m = getattr(s, m)
loopargs['output'][e][i]['start_time'] = time.time()
async with await m(url) as response:
ret = await response.text()
loopargs['output'][e][i]['end_time'] = time.time()
loopargs['output'][e][i]['status'] = response.status
# 细节:在每一个with前面加上async,在每一步的阻塞操作前加上await
except asyncio.TimeoutError:
print('time_out')
loopargs['output'][e][i]['time_out'] = 'time_out'
finally:
loopargs['input']['finished'] += 1
def sleep_to(t):
s = 0
while s < t:
s = time.time()
time.sleep(0.01)
def control_func(loop, loopargs):
# 添加任务
global stop_sign
while True:
e = time.time()
i = 0
ee = int(e)
loopargs['output'][ee] = {}
while i < loopargs['input']['times']:
loopargs['output'][ee][i] = {}
url = loopargs['input']['url']
asyncio.run_coroutine_threadsafe(
task(ee, i, 'get', url, headers, None, loopargs), loop)
loopargs['input']['started'] += 1
i += 1
print(ee, loopargs['input']['started'])
sleep_to(e+1)
if stop_sign:
while loopargs['input']['started'] != loopargs['input']['finished']:
time.sleep(1)
print( loopargs['input']['started'], loopargs['input']['finished'] )
asyncio.run_coroutine_threadsafe(task_stop(), loop)
break
print('ending...')
if __name__ == '__main__':
new_loop = asyncio.new_event_loop()
loopargs = {
'input':{
'times':500,
'url':'http://127.0.0.1:8080',
'started':0,
'finished':0,
},
'output':{
}
}
for sign_name in ('SIGINT', 'SIGTERM'):
new_loop.add_signal_handler(getattr(signal, sign_name),
functools.partial(exit_loop, sign_name))
t = Thread(target=control_func, args=(new_loop, loopargs) )
t.start()
asyncio.set_event_loop(new_loop)
new_loop.run_forever()
new_loop.close()
t.join()
# 归纳整理
for _, item in loopargs['output'].items():
ok = len(list(filter(lambda x: 'status' in x and x['status'] == 200 , item.values())))
all_res = len(list(filter(lambda x: 'status' in x , item.values())))
tt = len(list(filter(lambda x: 'time_out' in x, item.values())))
s = len(list(filter(lambda x: 'start_time' in x, item.values())))
print(f'{_}, 发送总数:{s}, 超时数量:{tt}, 返回数量:{all_res}, 成功数量:{ok}')
版权声明:除特别注明外,本站所有文章均为王晨曦个人站点原创
转载请注明:出处来自王晨曦个人站点 » 服务器性能测试客户端,aiohttp实现
出发地 [url=http://www.g65wn1w0bcl9k26491ut2c3z3a6itu71s.org/]usszhfokjq[/url] sszhfokjq http://www.g65wn1w0bcl9k26491ut2c3z3a6itu71s.org/ asszhfokjq