性能测试工具分布式压力测试代码篇

测试小书童 · 2016年12月28日 · 最后由心向东回复于 2016年12月30日 · 3967 次阅读

说明

主要用途是想用多机压测服务器端
可以组合任何平台，任意语言，任意压测方法，任意电脑等对服务器进行压测

设计思路

开启一个 webserver 记录客户端压测情况
然后可以运行多个电脑启动多个客户端，压测电脑在同一个内网，也可配置成外网 apache。

代码

配置信息

__author__ = "shikun"
class Const(object):
    log = "D:/app/Apache2.2/htdocs/client.log" # apache 的log路径
    APAHEC_IP = "192.168.1.38" # 本机ip
    PORT_NUMBER = 8088 # 端口号

web server.py

from http.server import BaseHTTPRequestHandler,HTTPServer
import urllib.parse
from common import operateFile
from common.customConst import Const
class myHandler(BaseHTTPRequestHandler):
  # Handler for the GET requests
    def do_GET(self):
        print('Get request received')
        req = urllib.parse.unquote(self.path)
        result = urllib.parse.parse_qs(req[2:]) # 得到请求参数
        self.send_response(200)
        self.send_header('Content-type','text/html')
        self.end_headers()
        # Send the html message
        self.wfile.write(b"ok!") #发送信息给客户端
        operateFile.write_txt(line=result["msg"][0], f_path=Const.log) # 记录各个客户端发来的信息
try:
  server = HTTPServer((Const.APAHEC_IP, Const.PORT_NUMBER), myHandler)
  print ('Started httpserver on port ' , Const.PORT_NUMBER)
  server.serve_forever()
except KeyboardInterrupt:
  print ('^C received, shutting down the web server')
  server.socket.close()

client.py

from multiprocessing import Process
from gevent import monkey; monkey.patch_all()
import gevent
import requests
import json
from common.customConst import Const
url = "http://rap.taobao.org/mockjsdata/10296/getUserInfo?id=2"
num = 10
result = {"success": 0, "computer": "压测客户端1", "cpu": 4, "men": 4, sum: 10} # 客户端信息

class Producer(object):
    '''
    协程发请求
    '''
    def __init__(self):
       self._rungevent()
       self.h = 0
    def _rungevent(self):
        jobs = []
        for i in range(num): #windows下有端口限制
            jobs.append(gevent.spawn(self.produce))
        gevent.joinall(jobs)
        requests.get("http://"+Const.APAHEC_IP+":"+ str(Const.PORT_NUMBER)+"/?msg="+str(result)) # 发送客户端的请求情况
    def produce(self):
        r = requests.get(url)
        if r.status_code == 200:
            r.encoding = 'UTF-8'
            if json.loads(r.text)["code"] == 0:
                result["success"] += 1
        else:
            print("失败咯")
def main():
    p = Process(target=Producer, args=())  # 一个进程启动协程
    p.start()

    # p1 = Process(target = Producer, args=())
    # p1.start()

if __name__ == '__main__':
    main()