python并发测试脚本

这两天要做性能测试，自己没事用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……

评论

3 楼 晴天i 2016-07-17  

写得真好，真好，配得注释也很好！！ 

2 楼 晴天i 2016-07-17  

能评论了吗

1 楼 u010530067 2015-11-06  

怎么运动你写的这个脚本？