Python 으로 구현 되 는 웹 서버 의 압력 테스트 도구

# coding:gbk
import time
import urllib2
import threading
from Queue import Queue
from time import sleep

#       
PERF_TEST_URL = "http://localhost:8080/perf.jsp"

#   :    
#THREAD_NUM = 10			#       
#ONE_WORKER_NUM = 500		#          
#LOOP_SLEEP = 0.01		#         ( )

#   :      
THREAD_NUM = 100		#       
ONE_WORKER_NUM = 10000		#          
LOOP_SLEEP = 0.5		#         ( )



#    
ERROR_NUM = 0


#       ，        
def doWork(index):
	t = threading.currentThread()
	#print "["+t.name+" "+str(index)+"] "+PERF_TEST_URL

	try:
		html = urllib2.urlopen(PERF_TEST_URL).read()
	except urllib2.URLError, e:
		print "["+t.name+" "+str(index)+"] "
		print e
		global ERROR_NUM
		ERROR_NUM += 1


#       ，             
def working():
	t = threading.currentThread()
	print "["+t.name+"] Sub Thread Begin"

	i = 0
	while i < ONE_WORKER_NUM:
		i += 1
		doWork(i)
		sleep(LOOP_SLEEP)

	print "["+t.name+"] Sub Thread End"


def main():
	#doWork(0)
	#return
	
	t1 = time.time()

	Threads = []

	#     
	for i in range(THREAD_NUM):
	    t = threading.Thread(target=working, name="T"+str(i))
	    t.setDaemon(True)
	    Threads.append(t)

	for t in Threads:
	    t.start()

	for t in Threads:
	    t.join()

	print "main thread end"

	t2 = time.time()
	print "========================================"
	print "URL:", PERF_TEST_URL
	print "    :", THREAD_NUM, "*", ONE_WORKER_NUM, "=", THREAD_NUM*ONE_WORKER_NUM
	print "   ( ):", t2-t1
	print "      ( ):", (t2-t1) / (THREAD_NUM*ONE_WORKER_NUM)
	print "       :", 1 / ((t2-t1) / (THREAD_NUM*ONE_WORKER_NUM))
	print "    :", ERROR_NUM


if __name__ == "__main__": main()