51Testing软件测试论坛

 找回密码
 (注-册)加入51Testing

QQ登录

只需一步,快速开始

微信登录,快人一步

手机号码,快捷登录

查看: 1463|回复: 2
打印 上一主题 下一主题

[原创] 分布式压力测试代码篇

[复制链接]
  • TA的每日心情
    无聊
    2024-7-12 13:16
  • 签到天数: 1 天

    连续签到: 1 天

    [LV.1]测试小兵

    跳转到指定楼层
    1#
    发表于 2017-6-26 14:05:35 | 只看该作者 回帖奖励 |倒序浏览 |阅读模式
    说明

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

    开启一个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} # 客户端信息

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

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

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

    结束语

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


    分享到:  QQ好友和群QQ好友和群 QQ空间QQ空间 腾讯微博腾讯微博 腾讯朋友腾讯朋友
    收藏收藏
    回复

    使用道具 举报

  • TA的每日心情

    1720761397
  • 签到天数: 1 天

    连续签到: 1 天

    2#
    发表于 2017-6-26 15:51:33 | 只看该作者
    你可以测试一下一次性压几百个用户看看是否支持
    回复 支持 反对

    使用道具 举报

    该用户从未签到

    3#
    发表于 2017-6-26 16:07:29 | 只看该作者
    交易量一般都是按百万起的,分布式的话 回收数据才是比较麻烦的功能点,我觉得你应该先玩下正规的压力测试工具lr或者jmter
    回复 支持 反对

    使用道具 举报

    本版积分规则

    关闭

    站长推荐上一条 /1 下一条

    小黑屋|手机版|Archiver|51Testing软件测试网 ( 沪ICP备05003035号 关于我们

    GMT+8, 2024-11-11 04:48 , Processed in 0.068143 second(s), 23 queries .

    Powered by Discuz! X3.2

    © 2001-2024 Comsenz Inc.

    快速回复 返回顶部 返回列表