python並發測試腳本

http://darklipeng.iteye.com/blog/1591753

這兩天要做性能測試，自己沒事用python寫了個腳本，用於壓力測試

 

# -*- coding: utf8 -*-
# code by Shurrik
import threading, time, httplib
HOST = "www.baidu.com"; #主機地址 例如192.168.1.101
PORT = 80 #端口
URI = "/?123" #相對地址,加參數防止緩存，否則可能會返回304
TOTAL = 0 #總數
SUCC = 0 #響應成功數
FAIL = 0 #響應失敗數
EXCEPT = 0 #響應異常數
MAXTIME=0 #最大響應時間
MINTIME=100 #最小響應時間，初始值為100秒
GT3=0 #統計3秒內響應的
LT3=0 #統計大於3秒響應的
# 創建一個 threading.Thread 的派生類
class RequestThread(threading.Thread):
    # 構造函數
    def __init__(self, thread_name):
        threading.Thread.__init__(self)
        self.test_count = 0

    # 線程運行的入口函數
    def run(self):

        self.test_performace()


    def test_performace(self):
            global TOTAL
            global SUCC
            global FAIL
            global EXCEPT
            global GT3
            global LT3
            try:
                st = time.time()
                conn = httplib.HTTPConnection(HOST, PORT, False)
                conn.request('GET', URI)
                res = conn.getresponse()
                #print 'version:', res.version
                #print 'reason:', res.reason
                #print 'status:', res.status
                #print 'msg:', res.msg
                #print 'headers:', res.getheaders()
                start_time
                if res.status == 200:
                    TOTAL+=1
                    SUCC+=1
                else:
                    TOTAL+=1
                    FAIL+=1
                time_span = time.time()-st
                print '%s:%f\n'%(self.name,time_span)
                self.maxtime(time_span)
                self.mintime(time_span)
                if time_span>3:
                    GT3+=1
                else:
                    LT3+=1
            except Exception,e:
                print e
                TOTAL+=1
                EXCEPT+=1
            conn.close()
    def maxtime(self,ts):
            global MAXTIME
            print ts
            if ts>MAXTIME:
                MAXTIME=ts
    def mintime(self,ts):
            global MINTIME
            if ts<MINTIME:
                MINTIME=ts

# main 代碼開始
print '===========task start==========='
# 開始的時間
start_time = time.time()
# 並發的線程數
thread_count = 300

i = 0
while i <= thread_count:
    t = RequestThread("thread" + str(i))
    t.start()
    i += 1
t=0
#並發數所有都完成或大於50秒就結束
while TOTAL<thread_count|t>50:
        print "total:%d,succ:%d,fail:%d,except:%d\n"%(TOTAL,SUCC,FAIL,EXCEPT)
        print HOST,URI
        t+=1
        time.sleep(1)
print '===========task end==========='
print "total:%d,succ:%d,fail:%d,except:%d"%(TOTAL,SUCC,FAIL,EXCEPT)
print 'response maxtime:',MAXTIME
print 'response mintime',MINTIME
print 'great than 3 seconds:%d,percent:%0.2f'%(GT3,float(GT3)/TOTAL)
print 'less than 3 seconds:%d,percent:%0.2f'%(LT3,float(LT3)/TOTAL)

 300並發試着測了iteye幾次，還是被封了一次IP……