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()