TA的每日心情 | 无聊 2024-7-12 13:16 |
---|
签到天数: 1 天 连续签到: 1 天 [LV.1]测试小兵
|
说明
主要用途是想用多机压测服务器端
可以组合任何平台,任意语言,任意压测方法,任意电脑等对服务器进行压测
设计思路
开启一个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()
复制代码 压测端可以用其他语言,其他方法如,多进程+线程,异步等
结果
结束语
这样算不算分布式压力测试,如果哪里有错误,欢迎指正。
服务器监控,现在做的比较简单.直接用命名即可。
|
|