本文實(shí)例為大家分享了python多線程http壓力測(cè)試的具體代碼,供大家參考,具體內(nèi)容如下
#coding=utf-8
import sys
import time
import thread
import httplib, urllib
import random
import uuid
import logging
logging.basicConfig(level=logging.DEBUG,
format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',
datefmt='%a, %d %b %Y %H:%M:%S',
filename='測(cè)試腳本日志.log',
filemode='w')
def log_uncaught_exceptions(exception_type, exception, tb):
logging.critical(''.join(traceback.format_tb(tb)))
logging.critical('{0}: {1}'.format(exception_type, exception))
sys.excepthook = log_uncaught_exceptions
#網(wǎng)關(guān)地址
addr="172.18.2.4"
port=8080
thread_count = 15 #單次并發(fā)數(shù)量
requst_interval = 10 #請(qǐng)求間隔(秒)
test_count = sys.maxsize #sys.maxsize # 指定測(cè)試次數(shù)
#字段說(shuō)明,必須一一對(duì)應(yīng)
#login為空表示使用隨機(jī)用戶名
param_list=[
{"login":"user1","password":"qweqwe12"},
]
now_count = 0
lock_obj = thread.allocate()
def send_http():
global now_count
httpClient = None
try:
for user in user_list:
tmp_user = user["login"]
if tmp_user.strip() =='':
tmp_user = str(uuid.uuid1()) + str(random.random())
print tmp_user
params = urllib.urlencode({"operationData":
[{"login": tmp_user,"password":user["password"]}]})
headers = {"Content-type": "application/x-www-form-urlencoded", "Accept": "text/plain"}
httpClient = httplib.HTTPConnection(addr, port, timeout=5)
httpClient.request("POST", "/simple/spider.task.distribute", params, headers)
response = httpClient.getresponse()
print '發(fā)送數(shù)據(jù): ' + params
print '返回碼: ' + str(response.status)
print '返回?cái)?shù)據(jù): ' + response.read()
logging.info('發(fā)送數(shù)據(jù): ' + params)
logging.info('返回碼: ' + str(response.status))
logging.info('返回?cái)?shù)據(jù): ' + response.read())
#print response.getheaders() #獲取頭信息
sys.stdout.flush()
now_count+=1
except Exception, e:
print e
logging.info(e)
finally:
if httpClient:
httpClient.close()
def test_func(run_count):
global now_count
global requst_interval
global lock_obj
cnt = 0
while cnt < run_count:
lock_obj.acquire()
print ''
print '***************************請(qǐng)求次數(shù):' + str(now_count) + '*******************************'
print 'Thread:(%d) Time:%s\n'%(thread.get_ident(), time.ctime())
logging.info(' ')
logging.info('***************************請(qǐng)求次數(shù):' + str(now_count) + '*******************************')
logging.info('Thread:(%d) Time:%s\n'%(thread.get_ident(), time.ctime()))
cnt+=1
send_http()
sys.stdout.flush()
lock_obj.release()
time.sleep(requst_interval)
def test(ct):
global thread_count
for i in range(thread_count):
thread.start_new_thread(test_func,(ct,))
if __name__=='__main__':
global test_count
test(test_count)
while True:
time.sleep(100)
以上就是本文的全部?jī)?nèi)容,希望對(duì)大家的學(xué)習(xí)有所幫助,也希望大家多多支持腳本之家。
更多文章、技術(shù)交流、商務(wù)合作、聯(lián)系博主
微信掃碼或搜索:z360901061
微信掃一掃加我為好友
QQ號(hào)聯(lián)系: 360901061
您的支持是博主寫(xiě)作最大的動(dòng)力,如果您喜歡我的文章,感覺(jué)我的文章對(duì)您有幫助,請(qǐng)用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧,狠狠點(diǎn)擊下面給點(diǎn)支持吧,站長(zhǎng)非常感激您!手機(jī)微信長(zhǎng)按不能支付解決辦法:請(qǐng)將微信支付二維碼保存到相冊(cè),切換到微信,然后點(diǎn)擊微信右上角掃一掃功能,選擇支付二維碼完成支付。
【本文對(duì)您有幫助就好】元

