python学习笔记（threading性能压力测试）

  1 #!/usr/bin/env python
  2 # -*- coding: utf_8 -*-
  3 
  4 import threading
  5 import requests
  6 import time
  7 import re
  8 from time import sleep
  9 
 10 # -------接口性能测试配置-------
 11 method = "post"
 12 # 接口类型
 13 url = "http://localhost:8081/swcw/back/sysLogin.action"
 14 # 接口地址
 15 data = {"username": "admin", "password": "123456"}
 16 # 接口参数
 17 thread_num = 20
 18 # 线程数
 19 one_work_num = 2
 20 # 每个线程循环次数
 21 loop_sleep = 1
 22 # 每次请求时间间隔
 23 response_time = []
 24 # 平均响应时间列表
 25 error = []
 26 # 错误信息列表
 27 
 28 
 29 class CreateThread:
 30     def __init__(self):
 31         pass
 32 
 33     @classmethod
 34     def thread_api(cls):
 35         global results
 36         try:
 37             if method == "post":
 38                 results = requests.post(url, data)
 39             if method == "get":
 40                 results = requests.get(url, data)
 41             return results
 42         except requests.ConnectionError:
 43             return results
 44     # 接口函数
 45 
 46     @classmethod
 47     def thread_response(cls):
 48         responsetime = float(CreateThread.thread_api().elapsed.microseconds) / 1000
 49         return responsetime
 50     # 获取响应时间 单位ms
 51 
 52     @classmethod
 53     def thread_response_avg(cls):
 54         avg = 0.000
 55         l = len(response_time)
 56         for num in response_time:
 57             avg += 1.000 * num / l
 58         return avg
 59     # 获取平均相应时间 单位ms
 60 
 61     @classmethod
 62     def thread_time(cls):
 63         return time.asctime(time.localtime(time.time()))
 64     # 获取当前时间格式
 65 
 66     @classmethod
 67     def thread_error(cls):
 68         try:
 69             pa = u"个人信息"
 70             pattern = re.compile(pa)
 71             match = pattern.search(CreateThread.thread_api().text)
 72             if CreateThread.thread_api().status_code == 200:
 73                 pass
 74                 if match.group() == pa:
 75                     pass
 76             else:
 77                 error.append(CreateThread.thread_api().status_code)
 78         except AttributeError:
 79             error.append("登录失败")
 80     # 获取错误的返回状态码
 81 
 82     @classmethod
 83     def thread_work(cls):
 84         threadname = threading.currentThread().getName()
 85         print "[", threadname, "] Sub Thread Begin"
 86         for i in range(one_work_num):
 87             CreateThread.thread_api()
 88             print "接口请求时间： ", CreateThread.thread_time()
 89             response_time.append(CreateThread.thread_response())
 90             CreateThread.thread_error()
 91             sleep(loop_sleep)
 92         print "[", threadname, "] Sub Thread End"
 93     # 工作线程循环
 94 
 95     @classmethod
 96     def thread_main(cls):
 97         start = time.time()
 98         threads = []
 99         for i in range(thread_num):
100             t = threading.Thread(target=CreateThread.thread_work())
101             t.setDaemon(True)
102             threads.append(t)
103         for t in threads:
104             t.start()
105         # 启动所有线程
106         for t in threads:
107             t.join()
108         # 主线程中等待所有子线程退出
109         end = time.time()
110 
111         print "========================================================================"
112         print "接口性能测试开始时间：", time.asctime(time.localtime(start))
113         print "接口性能测试结束时间：", time.asctime(time.localtime(end))
114         print "接口地址：", url
115         print "接口类型：", method
116         print "线程数：", thread_num
117         print "每个线程循环次数：", one_work_num
118         print "每次请求时间间隔：", loop_sleep
119         print "总请求数：", thread_num * one_work_num
120         print "错误请求数：", len(error)
121         print "总耗时（秒）：", end - start
122         print "每次请求耗时（秒）：", (end - start) / (thread_num * one_work_num)
123         print "每秒承载请求数（TPS)：", (thread_num * one_work_num) / (end - start)
124         print "平均响应时间（毫秒）：", CreateThread.thread_response_avg()
125 
126 
127 if __name__ == ‘__main__‘:
128     CreateThread.thread_main()