用Python實現的Web服務器的壓力測試工具

用Python寫了一個網頁抓取的工具，用來做網頁的服務器壓力測試。也可以模擬運行時的狀態。

1. # coding:gbk   
2. import time  
3. import urllib2  
4. import threading  
5. from Queue import Queue  
6. from time import sleep  
7.   
8. # 性能測試頁面   
9. PERF_TEST_URL = "http://localhost:8080/perf.jsp"  
10.   
11. # 配置:壓力測試   
12. #THREAD_NUM = 10            # 並發線程總數   
13. #ONE_WORKER_NUM = 500       # 每個線程的循環次數   
14. #LOOP_SLEEP = 0.01      # 每次請求時間間隔(秒)   
15.   
16. # 配置:模擬運行狀態   
17. THREAD_NUM = 100        # 並發線程總數   
18. ONE_WORKER_NUM = 10000      # 每個線程的循環次數   
19. LOOP_SLEEP = 0.5        # 每次請求時間間隔(秒)   
20.   
21.   
22.   
23. # 出錯數   
24. ERROR_NUM = 0  
25.   
26.   
27. #具體的處理函數，負責處理單個任務   
28. def doWork(index):  
29.     t = threading.currentThread()  
30.     #print "["+t.name+" "+str(index)+"] "+PERF_TEST_URL   
31.   
32.     try:  
33.         html = urllib2.urlopen(PERF_TEST_URL).read()  
34.     except urllib2.URLError, e:  
35.         print "["+t.name+" "+str(index)+"] "  
36.         print e  
37.         global ERROR_NUM  
38.         ERROR_NUM += 1  
39.   
40.   
41. #這個是工作進程，www.linuxidc.com負責不斷從隊列取數據並處理   
42. def working():  
43.     t = threading.currentThread()  
44.     print "["+t.name+"] Sub Thread Begin"  
45.   
46.     i = 0  
47.     while i < ONE_WORKER_NUM:  
48.         i += 1  
49.         doWork(i)  
50.         sleep(LOOP_SLEEP)  
51.   
52.     print "["+t.name+"] Sub Thread End"  
53.   
54.   
55. def main():  
56.     #doWork(0)   
57.     #return   
58.       
59.     t1 = time.time()  
60.   
61.     Threads = []  
62.   
63.     # 創建線程   
64.     for i in range(THREAD_NUM):  
65.         t = threading.Thread(target=working, name="T"+str(i))  
66.         t.setDaemon(True)  
67.         Threads.append(t)  
68.   
69.     for t in Threads:  
70.         t.start()  
71.   
72.     for t in Threads:  
73.         t.join()  
74.   
75.     print "main thread end"  
76.   
77.     t2 = time.time()  
78.     print "========================================"  
79.     print "URL:", PERF_TEST_URL  
80.     print "任務數量:", THREAD_NUM, "*", ONE_WORKER_NUM, "=", THREAD_NUM*ONE_WORKER_NUM  
81.     print "總耗時(秒):", t2-t1  
82.     print "每次請求耗時(秒):", (t2-t1) / (THREAD_NUM*ONE_WORKER_NUM)  
83.     print "每秒承載請求數:", 1 / ((t2-t1) / (THREAD_NUM*ONE_WORKER_NUM))  
84.     print "錯誤數量:", ERROR_NUM  
85.   
86.   
87. if __name__ == "__main__": main()