欢迎光临
梦想从学习开始!

分布式压力测试代码篇| 小熊测试

本文主要介绍 分布式压力测试代码篇| 小熊测试,小熊希望对大家的学习或者工作具有一定的参考学习价值,在测试领域有所提升和发展。

 
 说明

  · 主要用途是想用多机压测服务器端

  · 可以组合任何平台,任意语言,任意压测方法,任意电脑等对服务器进行压测

  
设计思路

  · 开启一个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()

  压测端可以用其他语言,其他方法如,多进程+线程,异步等

  结果

您现在正在阅读的是由小熊分享邦为您整理的 分布式压力测试代码篇| 小熊测试
  结束语

  这样算不算分布式压力测试,如果哪里有错误,欢迎指正。

  服务器监控,现在做的比较简单.直接用命名即可。也可以看此篇的压测统计思路

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持小熊分享邦(www.xxfxb.com),希望大家能坚持软件测试之路,谢谢。

赞(0) 打赏
未经允许不得转载:小熊分享邦 » 分布式压力测试代码篇| 小熊测试

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏