python接口多线程压测

　　工作中需要，编写一个python脚本放置在服务器上进行压测，压测利用多线程。第一次使用不是很理解，有不对的地方请指教。

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import requests, time, threading, random,os,codecs


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, press_url):
        self.press_url = press_url
        self.headers = self.headers


    def testinterface(self):
        '''压测接口'''
        data = {
            "taskDesc": "000000",
            "timeRange": [
                {
                    "startTime": "2021-01-05 15:00:00",
                    "endTime": "2021-12-21 23:59:59"
                }
            ],
            "processId": 1,
            "repeatTimes": 1,
            "jsonData": [
                {
                    "phoneNum1": str(140)+str(random.randint(10000000,99999999)),
                    "phoneNum2": "",
                    "phoneNum3": "",
                    "userData": "{"信用等级":"D6","地址":"01街","资金":"100","欠费月份":"2019-10-01"}",
                    "phoneNum4": "",
                    "phoneNum5": "",
                    "custId": "we_" + str(codecs.encode(os.urandom(8), 'hex').decode()),
                    "username": "小范"
                }
            ],
            "projectSn": "1048",
            "requestId": "wek_e00000004",
            "taskName": "we_e00000004",
            "enterpriseID": "200000110",
            "token": "C7AA14C74225CF84918F5B2DF8BBD4FB"
        }
        print(data)
        global ERROR_NUM
        try:
            reponse = requests.post(self.press_url, headers=self.headers,json=data)
            result = reponse.json()
            if result["code"] != 0:
                print(result)
                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.testinterface()
        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__':
    press_url = 'http://127.0.0.1:6005/xxxx/xxx/xxxxx'

    THREAD_NUM = 2  # 并发线程总数
    ONE_WORKER_NUM = 5  # 每个线程的循环次数
    LOOP_SLEEP = 0.01  # 每次请求时间间隔(秒)
    ERROR_NUM = 0  # 出错数

    obj = Presstest(press_url=press_url)
    obj.run()