python的一个接口压测脚本

 1 import requests  2 import queue  #Queue模块中提供了同步的、线程安全的队列类，包括 
 3                      #FIFO（先入先出)队列Queue，LIFO（后入先出）队列 
 4                      #LifoQueue，和优先级队列PriorityQueue。这些队列都 
 5                      #实现了锁原语，可在多线程通讯中直接使用。
 6 import threading  7 import time  8 
 9 status_code_list = [] 10 exec_time = 0 11 class MyThreadPool: 12     def __init__(self, maxsize): #定义队列时有一个默认的参数 
13         #maxsize, 若是不指定队列的长度，即manxsize=0,那么队列的长 
14       #度为无限长，若是定义了大于0的值，那么队列的长度就是maxsize。
15         self.maxsize = maxsize 16         self._pool = queue.Queue(maxsize) 17                   #maxsize设置队列的大小为pool的大小
18         for _ in range(maxsize):    #为何用一个下划线，由于实际上这 
19                     #里没用到这个变量，因此用一个符号就能够了。
20             self._pool.put(threading.Thread)    #往pool里放线程数
21 
22     def get_thread(self): 23         return self._pool.get() 24 
25     def add_thread(self): 26  self._pool.put(threading.Thread) 27 
28 def request_time(func): 29     def inner(*args, **kwargs): 30         global exec_time 31         start_time = time.time() 32         func(*args, **kwargs) 33         end_time = time.time() 34         exec_time = end_time-start_time 35 
36     return inner 37 
38 
39 def get_url(url): 40     global x,status_code_list 41     headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/65.0.3325.146 Safari/537.36', 42  } 43     response = requests.get(url,headers=headers) 44     code = response.status_code 45  status_code_list.append(code) 46     print(code) 47     return code 48 
49 
50 def get_count(_url='http://news.baidu.com/sports',_count=100):    # :param count: 每一个线程请求的数量
51     global status_code_list,url,count 52     for i in range(count): 53  get_url(url) 54 
55 def request_status(): 56     count_num = len(status_code_list) 57     set_code_list = set(status_code_list) 58     status_dict = {} 59     for i in set_code_list: 60         status_dict[i] = str(status_code_list).count(str(i)) 61  echo_str(count_num, set_code_list, status_dict) 62 
63 def echo_str(count_num,set_code_list,status_dict): 64     print('=======================================') 65     print('请求总次数:%s'%count_num) 66     print('请求时长:%s秒'%int(exec_time)) 67     second_request = count_num/int(exec_time) 68     print('每秒请求约:%s次'%int(second_request)) 69     print('状态码 | 次数') 70 
71     for k,v in status_dict.items(): 72         print(str(k)+' | '+str(v)) 73     print('=======================================') 74 
75 
76 @request_time 77 def run(url,thread_num=10,thread_pool=10): 78     '''
79  :param thread_num: 总共执行的线程数(总的请求数=总共执行的线程数*每一个线程循环请求的数量) 80  :param thread_pool: 线程池数量 81  :param url: 请求的域名地址 82     '''
83     global x,status_code_list 84     pool = MyThreadPool(thread_pool) 85     for i in range(thread_num): 86         t = pool.get_thread() 87         obj = t(target=get_count) 88  obj.start() 89  obj.join() 90 
91 
92 if __name__ == '__main__': 93     count = 10  #单个线程的请求数
94     url = 'http://baidu.com'
95     run(url,100,100) 96     request_status()