#!/usr/bin/env python
# -*- coding:utf-8 -*-
# 說明:
"""
api接口壓力測試
url = "http://gd.wangfanwifi.com:16621/api/userAuth?time=300&product=wxlogin&limit=10000&open_type=temp"
"""
# code by Shurrik
import threading, time, httplib

HOST = ‘gd.wangfanwifi.com‘ #主機
PORT = ‘16621‘ #端口
URI = "/api/userAuth?time=300&product=wxlogin&limit=10000&open_type=temp" #相對地址
 


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 (‘[返回內容]: %s‘% res.read())
            print (‘[http狀態碼]: %s‘ % res.status)
            print (‘\n‘)
            start_time
            if res.status == 200:
                TOTAL += 1
                SUCC += 1
            else:
                TOTAL += 1
                FAIL += 1
            time_span = time.time()-st
            print (‘線程: %s\t時間: %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 += c1
            EXCEPT += 1
        conn.close()
    def maxtime(self,ts):
        global MAXTIME
        # print ‘時間: %s‘ % ts
        if ts > MAXTIME:
            MAXTIME = ts
    def mintime(self,ts):
        global MINTIME
        if ts < MINTIME:
            MINTIME = ts

# main 代碼開始
if __name__ == "__main__":
    print (‘=‘*50+‘ api接口性能測試開始 ‘+‘=‘*50)
    # 開始的時間
    start_time = time.time()
    # 並發的線程數
    thread_count = 400
    print (‘並發數: %d‘ % thread_count)

    i = 0
    while i <= thread_count:
        t = RequestThread("thread" + str(i))
        t.start()
        i += 1

    t = 0
    #並發數所有都完成或大於60秒就結束
    while TOTAL < thread_count or t > 60:
        print (‘總數: %d\t成功: %d\t失敗: %d\t出錯: %d\n‘ % (TOTAL,SUCC,FAIL,EXCEPT))
        t += 1
        time.sleep(1)

    print (‘=‘*50+‘ api接口性能測試結束 ‘+‘=‘*50)
    print (‘api接口地址: http://%s:%s%s‘ % (HOST,PORT,URI))
    print (‘並發數: %d‘ % thread_count)
    print (‘總數: %d\t成功: %d\t失敗: %d\t出錯: %d‘ % (TOTAL,SUCC,FAIL,EXCEPT))
    print (‘成功率: %0.2f\t失敗率: %0.2f\t出錯率: %0.2f‘ % (float(SUCC)/float(TOTAL),float(FAIL)/float(TOTAL),float(EXCEPT)/float(TOTAL)))
    # print ‘最大響應時間: ‘,MAXTIME
    # print ‘最小響應時間: ‘,MINTIME
    # print ‘大於3秒的請求數: %d\t百分比: %0.2f‘ % (GT3,float(GT3)/TOTAL)
    # print ‘小於3秒的請求數: %d\t百分比: %0.2f‘ % (LT3,float(LT3)/TOTAL)
    print (‘=‘*50+‘ api接口性能測試結束 ‘+‘=‘*50)

模擬ab壓力測試