1. 前言

由于公司的一個項目是基于B/S架構(gòu)與WEB服務通信，使用XML數(shù)據(jù)作為通信數(shù)據(jù)，在添加新功能時，WEB端與客戶端分別由不同的部門負責，所以在WEB端功能實現(xiàn)過程中，需要自己發(fā)起請求測試，于是便選擇了使用Python編寫此腳本。另外由于此腳本最開始希望能在以后發(fā)展成具有壓力測試的功能，所以除了基本的訪問之外，添加了多線程請求。

整個腳本主要涉及到的關于Python的知識點包括：

基于urllib.request的Http訪問

多線程

類與方法的定義

全局變量的定義與使用

文件的讀取與寫入

……

2. 源碼與結(jié)果

整個程序包括Python源碼和配置文件，由于源碼中有相應的注釋，所以就直接貼源碼吧，如下：

# TradeWeb測試腳本import threading, time, http.client, urllib.request, os#import matplotlib.pyplot as pltURL = 'http://127.0.0.1:8888/XXXXXXXXX/httpXmlServlet' # 在配置文件中讀取，此處將無效TOTAL = 0; # 總數(shù)SUCC = 0; # 響應成功數(shù)量FAIL = 0; # 響應失敗數(shù)量EXCEPT = 0 # 響應異常數(shù) MAXTIME = 0 # 最大響應時間 MINTIME = 100 # 最小響應時間，初始值為100秒COUNT_TIME = 0 # 總時間THREAD_COUNT = 0 # 記錄線程數(shù)量CODE_MAP = {200:0, 301:0, 302:0, 304:0} # 狀態(tài)碼信息(部分)RESULT_FILE = 'tradeWebResult.xml'  # 輸出結(jié)果文件REQUEST_DATA_FILE = 'requestData.config'  # 數(shù)據(jù)文件DATA = '''請在tradeWebRequestData.config文件中配置'''TIME_LIST = [] # 記錄訪問時間#創(chuàng)建一個threading.Thread的派生類class RequestThread(threading.Thread): #構(gòu)造函數(shù) def __init__(self, thread_name): threading.Thread.__init__(self) self.test_count = 0; #線程運行的入口函數(shù) def run(self): global THREAD_COUNT THREAD_COUNT += 1 #print("Start the count of thread:%d" %(THREAD_COUNT)) self.testPerformace() #測試性能方法 def testPerformace(self): global TOTAL  global SUCC  global FAIL  global EXCEPT  global DATA global COUNT_TIME global CODE_MAP global URL try:  st = time.time() #記錄開始時間  start_time  cookies = urllib.request.HTTPCookieProcessor()  opener = urllib.request.build_opener(cookies)  resp = urllib.request.Request(url=URL,     headers={'Content-Type':'text/xml', 'Connection':'Keep-Alive'},     data=DATA.encode('gbk'))  respResult = opener.open(resp)  # 記錄狀態(tài)碼 START  code = respResult.getcode()  if code == 200:  SUCC += 1  else:  FAIL += 1  if code in CODE_MAP.keys():  CODE_MAP[code] += 1  else:  CODE_MAP[code] = 1  # print(request.status)  # 記錄狀態(tài)碼 END    html = respResult.read().decode('gbk')  print(html)  time_span = time.time() - st # 計算訪問時間  # 記錄訪問時間  TIME_LIST.append(round(time_span * 1000))  # print('%-13s: %f ' %(self.name, time_span))  self.maxtime(time_span)  self.mintime(time_span)  self.writeToFile(html)  # info = respResult.info() # 響應頭信息  # url = respResult.geturl() # URL地址  # print(info);  # print(url)  COUNT_TIME += time_span  TOTAL += 1 except Exception as e:  print(e)  TOTAL += 1  EXCEPT += 1 # 設置最大時間，如果傳入的時間大于當前最大時間 def maxtime(self, ts): global MAXTIME #print("time:%f" %(ts)) if ts > MAXTIME:  MAXTIME = ts # 設置最小時間，如果傳入的時間小于當前最小時間 def mintime(self, ts): global MINTIME #print("time:%f" %(ts)) if ts < MINTIME and ts > 0.000000000000000001:  MINTIME = ts # 寫入文件 def writeToFile(self, html): f = open(RESULT_FILE, 'w') f.write(html) f.write('/r/n') f.close();# 讀取XML數(shù)據(jù)信息def loadData(): global URL global DATA f = open(REQUEST_DATA_FILE, 'r') URL = "".join(f.readline()) DATA = "".join(f.readlines()) # print(DATA) f.close()if __name__ == "__main__": # print("============測試開始============") print("") # 開始時間 start_time = time.time() # 并發(fā)的線程數(shù) thread_count = 1 loadData() # 加載請求數(shù)據(jù) i = 0 while i < thread_count: t = RequestThread("Thread" + str(i)) t.start() i += 1 t = 0 while TOTAL < thread_count and t < 60: # print("total:%d, succ:%d, fail:%d, except:%d/n" %(TOTAL,SUCC,FAIL,EXCEPT)) print("正在請求 ",URL) t += 1 time.sleep(1) # 打印信息 print() print("請求", URL, "的統(tǒng)計信息：") print(" 總請求數(shù) = %d，成功 = %d，失敗 = %d，異常 = %d" %(TOTAL, SUCC, FAIL, EXCEPT)) print() print("往返程的估計時間(以毫秒為單位)：") print(" 合計 =", int(COUNT_TIME * 1000), "ms", end = '') print(" 最大 =", round(MAXTIME * 1000), "ms", end = '') print(" 最小 =", round(MINTIME * 1000), "ms", end = '') print(" 平均 =", round((COUNT_TIME / thread_count) * 1000), "ms") print() print("響應的狀態(tài)碼與次數(shù)信息(狀態(tài)碼:次數(shù))：") print(" ", CODE_MAP) print() print("輸出頁面請查看", RESULT_FILE, "文件(建議使用瀏覽器或XML專業(yè)工具打開)") print() # os.system("pause") print(TIME_LIST) input()