python自动化接口压测，用Thread模块进行接口多线程压测！！！

""" Thread 线程模块进行接口压测 """

import threading,time

class My_Thread:

   def __init__(self):

      # 添加总启动的线程列表 self.all_thread = []

      # 这里用 requests 中的elapsed.total_seconds()方法统计接口请求时间 self.requests_time = []

      # 添加成功的测试请求 self.Pass_requests = []

      # 添加失败的请求接口 self.Fail_requests = []

      def test_script(self,event,threadName,runTime):

        """ :param event:

        Thread类中的event方法 :

        param threadName:

        线程数 :param runTime:

        持续时间，分钟单位 """

        print("线程 {} 初始化完毕，随时可以启动...".format(threadName))

        # 线程等待

        event.wait()

        print("线程 {} 开始执行...".format(threadName))

        # 获取当前时间秒

        now = time.time()

        while True:

          if now + (runTime * 60) > time.time():

            """ 这里可以添加请求的接口/要测试的系统接口流程 """

            print("请求接口中...")

          else:

              break

          def run_thread(self,threadNum,runTime,startSeconds,endSeconds):

             """ :param threadNum 启动的线程数量

             :param runTime 持续时间分钟为单位

             :param startSeconds 每个多少秒启动一个

             :param endSeconds 每个多少秒结束一个 """

             # 实例化Event线程

             event = threading.Event()

             # 所有添加线程列表中

             for i in range(1,threadNum+1):

               self.all_thread.append(threading.Thread(target=self.test_script,args=(event,str(i),runTime)))

               event.clear()

                # 启动线程

               for thread in self.all_thread:

                   time.sleep(startSeconds)

                   thread.start()

                   event.set()

                   # 结束子线程

                   for thread in self.all_thread:

                     time.sleep(endSeconds)

                     thread.join()

                     def test_result(self):

                       """统计测试结果信息"""

                       print("==================== 测试结果数据 ====================")

                       print("总启动线程数：%s"%len(self.all_thread))

                       print("总耗时：%s"%sum(self.requests_time))

                       print("平均请求耗时：%s"%str(sum(self.requests_time)/len(self.requests_time)))

                       print("请求成功：%s"%len(self.Pass_requests))

                       print("请求失败：%s"%len(self.Fail_requests))

                       print("==================== 测试结果数据 ====================")

                       if __name__ == "__main__":

                         # 启动线程熟练 10 个

                         threadNum = 10

                         # 持续时间

                         runTime = 1

                         # 多长时间加载一个

                         startSeconds = 1

                         # 多长时间结束一个

                         endSeconds = 1

                         test = My_Thread().run_thread(threadNum,runTime,startSeconds,endSeconds)

#@title 3个colab thread = 50 + 1webbench999

"""

    利用threading模块进行接口压测

"""

import requests, threading, re, time, datetime

event = threading.Event()

def get_session():

    headers = {

        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",

        "Cookie": "MSO=SID&1579350070",

        "Referer": "keep-alive",

        "Connection": "keep-alive",

    }

    try:

        r = requests.get(url="http://www.hnyanling.gov.cn", headers=headers)

        regex = re.compile(r"name=userSession value=(.*?)>")

        return regex.findall(r.text)[0]

    except Exception as e:

        print(e)

request_tm = []

Pass = []

Fail = []

def login(running_time,event,i):

    headers = {

        "Content-Type": "application/x-www-form-urlencoded",

        "Cookie": "MSO=SID&1579351869",

        "Content-Length": "128"

    }

    body = {

        "userSession": get_session(),

        "username": "13720673941",

        "password": "111111",

        "login.x": "61",

        "login.y": "7",

        "JSFormSubmit": "off"

    }

    print("线程 {} 初始化完毕，随时可以启动...".format(i))

    # 线程等待

    event.wait()

    print("线程 {} 开始执行...".format(i))

    # 获取当前时间秒

    now = time.time()

    while True:

        if now + (running_time * 6660) > time.time():

            try:

                global r

                r = requests.get(url="http://www.hnyanling.gov.cn", headers=headers, data=body, timeout=15)

                request_tm.append(r.elapsed.total_seconds())

                if r.status_code == 200:

                    Pass.append(r.status_code)

                    print("第 {thread_num} 线程请求成功！协议状态码：{status_code}".format(thread_num=i, status_code=r.status_code))

            except Exception as e:

                Fail.append(r.status_code)

                print("第 {thread_num} 线程请求异常！协议状态码：{status_code}".format(thread_num=i, status_code=r.status_code))

                print(e)

        else:

            break

thread_list = []

def run(running_time,thread_num,start_time, end_time):

    """

    running_time: 持续时间 分钟为单位

    thread_num: 线程数

    start_time：几秒启动一个线程

    end_time：  几秒结束一个线程

    """

    for i in range(1, thread_num + 1):

        thread_list.append(threading.Thread(target=login, args=(running_time,event,i)))

    event.clear()

    for thread in thread_list:

        time.sleep(start_time)

        thread.start()

    event.set()

    for thread in thread_list:

        time.sleep(end_time)

        thread.join()

# --------------------------------------------------------------------------------

# 持续时间分钟

running_time = 1

# 线程数量

thread_num = 100

# 2秒增加一个线程

start_time = 1

# 1秒结束一个线程

end_time = 1

run(running_time, thread_num, start_time, end_time)

print("=================== 测试结果 ====================")

print("启动总线程：%s" % len(thread_list))

print("总耗时：%s" % sum(request_tm))

print("平均请求耗时：%s" % str(sum(request_tm) / len(request_tm)))

print("请求成功：%s" % len(Pass))

print("请求失败：%s" % len(Fail))

print("=================== 测试结束 ====================")