您还未登录! 登录 | 注册 | 帮助  

您的位置: 首页 > 软件测试技术 > 性能测试 > 正文

分布式压力测试代码篇

发表于:2017-01-09 作者:lose 来源:

  说明
  · 主要用途是想用多机压测服务器端
  · 可以组合任何平台,任意语言,任意压测方法,任意电脑等对服务器进行压测
  设计思路
  · 开启一个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()
  压测端可以用其他语言,其他方法如,多进程+线程,异步等
  结果

  结束语
  这样算不算分布式压力测试,如果哪里有错误,欢迎指正。
  服务器监控,现在做的比较简单.直接用命名即可。也可以看此篇的压测统计思路