【代碼片斷】利用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() 是阻塞等待全部進程執行結束。