51Testing软件测试论坛

标题: 分布式压力测试代码篇 [打印本页]
作者: 乐哈哈yoyo    时间: 2017-6-26 14:05
标题: 分布式压力测试代码篇
说明

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

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

代码
配置信息
  1. __author__ = "shikun"
  2. class Const(object):
  3.     log = "D:/app/Apache2.2/htdocs/client.log" # apache 的log路径
  4.     APAHEC_IP = "192.168.1.38" # 本机ip
  5.     PORT_NUMBER = 8088 # 端口号
复制代码
web server.py
  1. from http.server import BaseHTTPRequestHandler,HTTPServer
  2. import urllib.parse
  3. from common import operateFile
  4. from common.customConst import Const
  5. class myHandler(BaseHTTPRequestHandler):
  6.   # Handler for the GET requests
  7.     def do_GET(self):
  8.         print('Get request received')
  9.         req = urllib.parse.unquote(self.path)
  10.         result = urllib.parse.parse_qs(req[2:]) # 得到请求参数
  11.         self.send_response(200)
  12.         self.send_header('Content-type','text/html')
  13.         self.end_headers()
  14.         # Send the html message
  15.         self.wfile.write(b"ok!") #发送信息给客户端
  16.         operateFile.write_txt(line=result["msg"][0], f_path=Const.log) # 记录各个客户端发来的信息
  17. try:
  18.   server = HTTPServer((Const.APAHEC_IP, Const.PORT_NUMBER), myHandler)
  19.   print ('Started httpserver on port ' , Const.PORT_NUMBER)
  20.   server.serve_forever()
  21. except KeyboardInterrupt:
  22.   print ('^C received, shutting down the web server')
  23.   server.socket.close()
复制代码
client.py
  1. from multiprocessing import Process
  2. from gevent import monkey; monkey.patch_all()
  3. import gevent
  4. import requests
  5. import json
  6. from common.customConst import Const
  7. url = "http://rap.taobao.org/mockjsdata/10296/getUserInfo?id=2"
  8. num = 10
  9. result = {"success": 0, "computer": "压测客户端1", "cpu": 4, "men": 4, sum: 10} # 客户端信息

  11. class Producer(object):
  12.     '''
  13.     协程发请求
  14.     '''
  15.     def __init__(self):
  16.        self._rungevent()
  17.        self.h = 0
  18.     def _rungevent(self):
  19.         jobs = []
  20.         for i in range(num): #windows下有端口限制
  21.             jobs.append(gevent.spawn(self.produce))
  22.         gevent.joinall(jobs)
  23.         requests.get("http://"+Const.APAHEC_IP+":"+ str(Const.PORT_NUMBER)+"/?msg="+str(result)) # 发送客户端的请求情况
  24.     def produce(self):
  25.         r = requests.get(url)
  26.         if r.status_code == 200:
  27.             r.encoding = 'UTF-8'
  28.             if json.loads(r.text)["code"] == 0:
  29.                 result["success"] += 1
  30.         else:
  31.             print("失败咯")
  32. def main():
  33.     p = Process(target=Producer, args=())  # 一个进程启动协程
  34.     p.start()

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

  39. if __name__ == '__main__':
  40.     main()
复制代码
压测端可以用其他语言，其他方法如，多进程+线程，异步等
结果

结束语

这样算不算分布式压力测试，如果哪里有错误，欢迎指正。
服务器监控，现在做的比较简单.直接用命名即可。


作者: 悠悠小仙仙    时间: 2017-6-26 15:51
你可以测试一下一次性压几百个用户看看是否支持
作者: 草帽路飞UU    时间: 2017-6-26 16:07
交易量一般都是按百万起的，分布式的话 回收数据才是比较麻烦的功能点，我觉得你应该先玩下正规的压力测试工具lr或者jmter



欢迎光临 51Testing软件测试论坛 (http://bbs.51testing.com/) Powered by Discuz! X3.2