用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.bkjia.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()  

相关内容