Welcome 微信登录

首页 / 操作系统 / Linux / 用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（）  
版权所有©石家庄振强科技有限公司2024 冀ICP备08103738号-5 网站地图