Advertisement

手把手教你用python压测

阅读量:

手把手教你用python压测

Thread 线程模块进行接口压测

相关代码

添加总启动的线程列表

self.all_thread = []
这里用 requests 中的elapsed.total_seconds()方法统计接口请求时间
self.requests_time = []
添加成功的测试请求
self.Pass_requests = []
添加失败的请求接口
self.Fail_requests = []

事件处理函数:Thread类中的event方法;表示当前运行的线程数量:threadName;持续时间为运行时长(单位为分钟):runTime

复制代码
 """

    
     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("线程 {} 初始化完毕,随时可以启动...\n".format(threadName))
    
         # 线程等待
    
         event.wait()
    
         print("线程 {} 开始执行...\n".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)

全部评论 (0)

还没有任何评论哟~