背景說明
在工作中,要對一個接口進行壓測,我當時就想通過python自己編寫一個壓力發生器。
初步方案(單線程循環發送)
通過循環向服務端發送請求,代碼如下:
#采用單步循環的方式循環測試
import requests,time
def run(runnum):
url = "https://api-test.peanut.ai/wechatGrant/load/test1?openId=RP0ulQ4pHDTBWt77ILCs02QGU&bsscode=8167871547864571"
for i in range(runnum):
str_res = requests.get(url)
if __name__ == "__main__":
start_time = time.time()
run(100)
end_time = time.time()
#print("循環次數:",str(counut))
print("開始時間:",str(start_time))
print("結束時間:",str(end_time))
print("運行時間:",str(end_time - start_time))
測試結果如下:
|
單線程 |
1 |
開始時間: 1536545804.5258229 |
2 |
開始時間: 1536546027.6947124 |
3 |
開始時間: 1536546205.2600951 |
4 |
開始時間: 1536546368.2361982 |
5 |
開始時間: 1536546640.4913867 |
運行時間很長,對程序進行了分析,因為循環是單線程並且是同步的,發送請求后,必須等待收到響應,才會發送下一個請求,效率很低,並且循環對壓力機的CPU資源消耗較大。
多線程方案
考慮通過多線程提高測試效率,代碼如下:
import threading
import requests
import time
url = "https://api-test.peanut.ai/wechatGrant/load/test1?openId=RP0ulQ4pHDTBWt77ILCs02QGU&bsscode=8167871547864571"
def run_thread(snum,enum):
for i in range(snum,enum):
#s = requests.session()
#關閉長連接
headers = {'Connection': 'close'}
str_res = requests.get(url,headers=headers)
#從發送請求到收到響應消耗的時間,單位微妙。
etime = str_res.elapsed.microseconds/1000000
threads = []
for i in range(0,10):
#循環生成線程
t = threading.Thread(target=run_thread, args=(0,10))
threads.append(t)
if __name__ == "__main__":
start_time = time.time()
for dd in range(0,10):
#啟動線程
threads[dd].start()
for dd in range(0,10):
threads[dd].join()
end_time = time.time()
print("開始時間:",str(start_time))
print("結束時間:",str(end_time))
print("運行時間:",str(end_time - start_time))
測試結果如下:
|
單線程 |
多線程(10) |
1 |
開始時間: 1536545804.5258229 |
開始時間: 1536546926.9662883 |
2 |
開始時間: 1536546027.6947124 |
開始時間: 1536546962.368912 |
3 |
開始時間: 1536546205.2600951 |
開始時間: 1536546990.911677 |
4 |
開始時間: 1536546368.2361982 |
開始時間: 1536547021.2030604 |
5 |
開始時間: 1536546640.4913867 |
開始時間: 1536547046.3298163 |
測試效率有很大提高,但也存在問題,當啟動線程較多時,壓力機資源消耗大,在同一個線程內部,還是同步進行,效率較低。
異步通訊方案
asyncio可以實現單線程並發IO操作。如果僅用在客戶端,發揮的威力不大。如果把asyncio用在服務器端,例如Web服務器,由於HTTP連接就是IO操作,因此可以用單線程+coroutine實現多用戶的高並發支持。
#采用異步通訊的方式發壓
import requests,time
import asyncio
from aiohttp import ClientSession
import aiohttp
url = "https://api-test.peanut.ai/wechatGrant/load/test1?openId=RP0ulQ4pHDTBWt77ILCs02QGU&bsscode=8167871547864571"
tasks = []
async def run():
async with ClientSession() as session:
async with session.get(url) as response:
response = await response.read()
#print(response)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
start_time = time.time()
for i in range(100):
tasks.append(run())
end_time = time.time()
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
#print("循環次數:",str(counut))
print("開始時間:",str(start_time))
print("結束時間:",str(end_time))
print("運行時間:",str(end_time - start_time))
測試結果:
|
單線程 |
多線程(10) |
異步 |
1 |
開始時間: 1536545804.5258229 |
開始時間: 1536546926.9662883 |
開始時間: 1536565502.732898 |
2 |
開始時間: 1536546027.6947124 |
開始時間: 1536546962.368912 |
開始時間: 1536565502.732898 |
3 |
開始時間: 1536546205.2600951 |
開始時間: 1536546990.911677 |
開始時間: 1536565502.732898 |
4 |
開始時間: 1536546368.2361982 |
開始時間: 1536547021.2030604 |
開始時間: 1536565502.732898 |
5 |
開始時間: 1536546640.4913867 |
開始時間: 1536547046.3298163 |
開始時間: 1536565502.732898 |
性能大大提高,但還有一個需要優化的地方,tasks采用的是list,如果數量多了,會占用大量內存,下步進行優化。