用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
8. # 性能測試頁面
9. PERF_TEST_URL = "http://localhost:8080/perf.jsp"
11. # 配置:壓力測試
12. #THREAD_NUM = 10 # 並發線程總數
13. #ONE_WORKER_NUM = 500 # 每個線程的循環次數
14. #LOOP_SLEEP = 0.01 # 每次請求時間間隔(秒)
16. # 配置:模擬運行狀態
17. THREAD_NUM = 100 # 並發線程總數
18. ONE_WORKER_NUM = 10000 # 每個線程的循環次數
19. LOOP_SLEEP = 0.5 # 每次請求時間間隔(秒)
23. # 出錯數
24. ERROR_NUM = 0
27. #具體的處理函數，負責處理單個任務
28. def doWork(index):
29. t = threading.currentThread()
30. #print "["+t.name+" "+str(index)+"] "+PERF_TEST_URL
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
41. #這個是工作進程，www.linuxidc.com負責不斷從隊列取數據並處理
42. def working():
43. t = threading.currentThread()
44. print "["+t.name+"] Sub Thread Begin"
46. i = 0
47. while i < ONE_WORKER_NUM:
48. i += 1
49. doWork(i)
50. sleep(LOOP_SLEEP)
52. print "["+t.name+"] Sub Thread End"
55. def main():
56. #doWork(0)
57. #return
59. t1 = time.time()
61. Threads = []
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)
69. for t in Threads:
70. t.start()
72. for t in Threads:
73. t.join()
75. print "main thread end"
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
87. if __name__ == "__main__": main()