用 wrk 验证一下测试服务性能：(抛开带宽限制)
曾做过一个项目的全链路压测，目的是验证服务有 LB 层、虚拟化层再到服务本身的性能情况，目标 QPS 是 50w；这就带来一个问题，由于待测服务 A 有 n 个后端服务，且客户端请求数:后端服务请求数是 1:n 的，也就是说预期的 50w 的 qps，对应后端的 qps 和就是 n 倍的 50w；当然，测试过程不能使用线上服务，所以需要一个性能非常好的 mock 服务；
更多内容可以学习《测试工程师 Python 工具开发实战》书籍、《大话性能测试 JMeter 实战》书籍
需求 1：单机 100w 的 qps——使用 japronto 服务可实;
需求 2：自定义 mock 接口 path——服务启动时读取 DB 内配置作为 path;
需求 3：更新接口返回内容方便——前端页面支持增加&修改 DB 接口 path 配置;
需求 4：支持 json 结果&protobuf 二进制结果——预置 proto，自定义接口返回类型，服务初始化时做不同处理；
服务启动 -> initData -> 根据 type 做不同处理，默认返回类型 json，proto 的话做二进制序列化；

用 wrk 验证一下测试服务性能：(抛开带宽限制)
32 核机器，20 个进程，qps 可达 98w，tp99<2ms;
实际使用时，只需要同时部署 5 台 20 核虚机，或打包成镜像附属到测试集群，使用内网域名负载均衡到这五台机器（k8s 集群的话使用域名映射），将被测服务的依赖服务地址改写成 mock 域名:端口/自定义 path；
整个工程 100 行左右代码，比较简陋，前端页面在另一个内部工程中完成，不好贴出来，大家可以自己实现，或者直接通过 mysql 客户端修改 DB 内容；
目前 mock 服务只支持单一接口对应单一 mock 结果，后续可以自定义多个 value，在服务内随机返回；
1、fastMock.py
from japronto import Application
import etcd3, json, sys
import mysqlConnect
def handler(request):
# print(str(res[request.path]))
if(alldata[request.path]["type"]=="proto"):
return request.Response(body=alldata[request.path]["mockvalue"])
else:
r = json.dumps(alldata[request.path]["mockvalue"])
return request.Response(text=r)
def covProto(jsonstr):
#需要protobuf类型的response可以通过配置mock类型（proto）来返回二进制body
# serverResponseBody = json_format.Parse(jsonstr, prototest_pb2.ServerResponseBody())
# body = serverResponseBody.SerializeToString()
#return body
return "binnary body"
def initData():
mc = mysqlConnect.MysqlConnect({"HOST": "xx.xx.xx.xx", "USER": "root", "PWD": "xxxx", "DB": "mock"})
i = mc.query("select * from mockapi")
res = {}
for row in i:
if(row[3]=="proto"):
body = covProto(row[2])
res[row[1]] = {"type": row[3], "mockvalue": body}
else:
res[row[1]] = {"type": row[3], "mockvalue": row[2]}
return res
if __name__ == '__main__':
try:
theads = sys.argv[1]
except Exception as e:
theads = 1
print(sys.argv)
print(e)
app = Application()
alldata = initData()
for k in alldata:
print(k)
app.router.add_route(k, handler)
# worker_num 启动时设置，建议等于真cpu个数，debug=True时，控制台会输出请求的path
app.run(port=80, worker_num=int(theads), debug=False)
2、mysqlConnect.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import pymysql
class MysqlConnect():
def __init__(self, dbinfo={}):
conn = pymysql.connect(host=dbinfo["HOST"],
port=3306,
user=dbinfo["USER"],
password=dbinfo["PWD"],
db=dbinfo["DB"],
charset='utf8')
self.conn = conn
def query(self, sql):
cursor = self.conn.cursor()
cursor.execute(sql)
return cursor.fetchall()
def excute(self, sql, args):
cursor = self.conn.cursor()
# info = {'name': 'fake', 'age': 15}
effect_row = cursor.execute(sql, args)
self.conn.commit()
return effect_row
def __del__(self):
self.conn.close()
超高并发的中转服务测试场景，依赖服务数据稳定，被测服务无缓存（或测试过程中去掉缓存）
更多内容可以学习《测试工程师 Python 工具开发实战》书籍、《大话性能测试 JMeter 实战》书籍
