#coding=utf-8
import urllib2
import threading
import time
TOTAL = 0 #總數
SUCC = 0 #響應成功數
FAIL = 0 #響應失敗數
EXCEPT = 0 #響應異常數
MAXTIME=0 #最大響應時間
MINTIME=100 #最小響應時間，初始值為100秒
# 子類化Thread
class Mythread(threading.Thread):
def __init__(self, func, args, name=''):
threading.Thread.__init__(self)
self.name = name
self.func = func
self.args = args
def getResult(self):
return self.res
def run(self):
self.res = apply(self.func, self.args)
def request_url(url, r):
global TOTAL
global SUCC
global FAIL
global EXCEPT
try:
st = time.time()
res = urllib2.urlopen(url)
status = res.getcode()
if status == 200:
TOTAL+=1
SUCC+=1
else:
TOTAL+=1
FAIL+=1
time_span = time.time()-st
maxtime(time_span)
self.mintime(time_span)
except Exception, e:
TOTAL+=1
EXCEPT+=1
def maxtime(ts):
global MAXTIME
if ts>MAXTIME:
MAXTIME=ts
def mintime(ts):
global MINTIME
if ts<MINTIME:
MINTIME=ts
def main():
print '===========task start==========='
# 開始的時間
start_time = time.time()
# 並發的線程數
thread_count = 100
i = 0
while i <= thread_count:
t = Mythread(request_url, ("http://www.baidu.com", "x"))
t.start()
i += 1
t=0
#並發數所有都完成或大於20秒就結束
while TOTAL<thread_count|t>2:
print "total:%d,succ:%d,fail:%d,except:%d\n"%(TOTAL,SUCC,FAIL,EXCEPT)
t+=1
time.sleep(1)
print '===========task end==========='
print "total:%d,succ:%d,fail:%d,except:%d"%(TOTAL,SUCC,FAIL,EXCEPT)
print 'response maxtime:',MAXTIME
print 'response mintime',MINTIME
s = raw_input("Press any key")
print "bay!"
pass
if __name__ == "__main__":
main()
文章內容來源：
http://www.runoob.com/python/python-multithreading.html