近在編寫一個簡單的WEB服務(wù)器，一個日常的工作是測試服務(wù)器的性能，試用了MS的Web Application Stress，發(fā)現(xiàn)它居然不支持除80以外端口的測試，其他的如Loadrunner 太貴而且太大，試用版只支持10個并發(fā)用戶，我Google到了100個并發(fā)用戶的許可想試用一下，不過沒有安裝成功。想想這種壓力測試實(shí)際上沒啥技術(shù)含量，自己用Python來編寫了小段測試代碼。
使用Python的原因
毫無疑問，編寫這樣的代碼使用Python合適，使用C/C++編寫有點(diǎn)小題大做，使用C#編寫編譯又很麻煩，我是使用Editplus來寫代碼的，因?yàn)橐紤]做不同的測試，只能邊寫邊調(diào)整，使用Python，下載一個Python的加亮文件，設(shè)置User Tool 1 到 Python，運(yùn)行的時(shí)候只需要按Ctrl+1，著實(shí)方便的很。
壓力測試的實(shí)現(xiàn)
壓力測試是通過模擬對WEB服務(wù)器的訪問，進(jìn)行記錄響應(yīng)時(shí)間，計(jì)算每秒處理數(shù)，記錄上傳下載的字節(jié)數(shù)。一般來說，一臺測試機(jī)器上視機(jī)器的性能，發(fā)起 50～200的連接，基本差不多了�？紤]到測試機(jī)器的負(fù)載，一般使用多線程來完成多個WEB請求，幸運(yùn)的是，Python對所有的這些支持的相當(dāng)完善。以下是測試的代碼
# code by 李嘉
# 禁止任何商業(yè)目的的轉(zhuǎn)載
# 不對因使用代碼產(chǎn)生任何后果負(fù)任何責(zé)任
# 轉(zhuǎn)載請保留所有聲明
import threading， time， httplib， random
# 需要測試的 url 列表，每一次的訪問，我們隨機(jī)取一個
urls = [
"/test?page="，
"/test2?orderby=a&page="，
"/test2?orderby=d&page="，
]
MAX_PAGE = 10000
SERVER_NAME = "192.168.0.64:80"
TEST_COUNT = 10000
# 創(chuàng)建一個 threading.Thread 的派生類
class RequestThread(threading.Thread):
# 構(gòu)造函數(shù)
def __init__(self， thread_name):
threading.Thread.__init__(self)
self.test_count = 0
# 線程運(yùn)行的入口函數(shù)
def run(self):
# 不直接把代碼寫在run里面是因?yàn)橐苍S我們還要做其他形式的測試
i = 0
while i < TEST_COUNT:
self.test_performace()
i += 1
#self.test_other_things()
def test_performace(self):
conn = httplib.HTTPConnection(SERVER_NAME)
# 模擬 Keep-Alive 的訪問， HTTP 1.1
for i in range(0， random.randint(0， 100)):
# 構(gòu)造一個 url，提供隨機(jī)參數(shù)的能力
url = urls[random.randint(0， len(urls) - 1)];
url += str(random.randint(0， MAX_PAGE))
# 這連接到服務(wù)器上去
#print url
try:
conn.request("GET"， url)
rsps = conn.getresponse()
if rsps.status == 200:
# 讀取返回的數(shù)據(jù)
data = rsps.read()
self.test_count += 1
except:
continue
conn.close()
# main 代碼開始
# 開始的時(shí)間
start_time = time.time()
threads = []
# 并發(fā)的線程數(shù)
thread_count = 100
i = 0
while i < thread_count:
t = RequestThread("thread" + str(i))
threads.append(t)
t.start()
i += 1
# 接受統(tǒng)計(jì)的命令
word = ""
while True:
word = raw_input("cmd:")
if word == "s":
time_span = time.time() - start_time
all_count = 0
for t in threads:
all_count += t.test_count
print "%s Request/Second" % str(all_count / time_span)
elif word == "e":
# 準(zhǔn)備退出 其實(shí) X 掉 窗口更加容易，沒什么浪費(fèi)的資源
TEST_COUNT = 0
for t in threads:
t.join(0)
break
