本文实例为大家分享了python多线程http压力测试的具体代码,供大家参考,具体内容如下
#=utf - 8编码
导入系统
导入的时间
进口的线程
,的进口httplib的urllib
进口随机
进口uuid
导入日志=logging.DEBUG logging.basicConfig(水平,
格式=' % (asctime) %(文件名)年代[:% (lineno) d] % (levelname) %(消息)年代',
datefmt=' %, % d % b % Y % H: % M: % S的,
文件名='测试脚本日志. log”,
filemode=' w ')
def log_uncaught_exceptions (exception_type、异常、结核病):
. join (traceback.format_tb logging.critical (“(tb)))
日志记录。关键({0}:{1}。格式(exception_type除外))
sys。excepthook=log_uncaught_exceptions
#网关地址
addr=" 172.18.2.4 "
端口=8080
thread_count=15 #单次并发数量
requst_interval=10 #请求间隔(秒)
test_count=系统。最大尺寸#系统。最大尺寸#指定测试次数
#字段说明,必须一一对应
#登录为空表示使用随机用户名
param_list=[
{“登录”:“user1”,“密码”:“qweqwe12”},
]
now_count=0
lock_obj=thread.allocate ()
def send_http ():
全球now_count
httpClient=没有
试一试:
在user_list用户:
tmp_user=用户(“登录”)
如果tmp_user.strip ()==?
tmp_user=str (uuid.uuid1 ()) + str (random.random ())
打印tmp_user
params=urllib.urlencode ({“operationData”:
[{“登录”:tmp_user,“密码”:用户(“密码”)}]})
头={“内容类型”:“应用程序/x-www-form-urlencoded”,“接受”:“text/plain”}
httpClient。的=httplib的HTTPConnection (addr,港口,超时=5)
httpClient。请求(“文章”、“/简单/spider.task。分发”,params,标题)
响应=httpClient.getresponse ()
打印的发送数据:“+参数
打印“返回码:' + str (response.status)
打印“返回数据:' + response.read ()
logging.info(“发送数据:“+ params)
logging.info(“返回码:' + str (response.status))
logging.info(“返回数据:' + response.read ())
#打印response.getheaders() #获取头信息
sys.stdout.flush ()
now_count +=1
除了例外,e:
打印e
logging.info (e)
最后:
如果httpClient:
httpClient.close ()
def test_func (run_count):
全球now_count
全球requst_interval
全球lock_obj
问=0
而问& lt;run_count:
lock_obj.acquire ()
打印”
打印的* * * * * * * * * * * * * * * * * * * * * * * * * * *请求次数:' + str (now_count) +“* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *’
打印的线程:(% d)时间:% s \ n ' % (thread.get_ident (), time.ctime ())
logging.info (' ')
logging.info(“* * * * * * * * * * * * * * * * * * * * * * * * * * *请求次数:' + str (now_count) +“* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *’)
logging.info(“线程:(% d)时间:% s \ n ' % (thread.get_ident (), time.ctime ()))
问+=1
send_http ()
sys.stdout.flush ()
lock_obj.release ()
time . sleep (requst_interval)
def测试(ct):
全球thread_count
因为我在范围(thread_count):
thread.start_new_thread (test_func (ct))
if __name__==癬_main__”:
全球test_count
测试(test_count)
而真正的:
time . sleep (100)
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
