最近公司需要對(duì)服務(wù)進(jìn)行性能測(cè)試确虱，但是服務(wù)不開放http服務(wù)含友，都是rpc接口，如果用httprpc開放后校辩，再進(jìn)行測(cè)試窘问，必然會(huì)產(chǎn)生性能損耗，測(cè)試結(jié)果不準(zhǔn)確宜咒，那么就需要直接對(duì)grpc接口進(jìn)行性能測(cè)試惠赫，那現(xiàn)在我們開始吧！
本篇文章建立在故黑，你已經(jīng)編寫了觸發(fā)grpc的方法
底下的run方法是我已經(jīng)寫好的觸發(fā)rpc的方法
# -*- coding: utf-8 -*-
import time
from util import run
from locustimport events, User
class GrpcClient(object):
"""重寫self.client"""
_locust_environment =None
def connect(self, name, grpc_env, path, parms):
"""重寫connect方法"""
start_time =int(time.time())
try:
# 記錄開始時(shí)間
env = get_env(grpc_env)
res = run(env, path, parms)
except Exception as e:
total_time =int((time.time() - start_time) *1000)
events.request_failure.fire(
request_type='grpc',
name=name,
response_time=total_time,
exception=e
)
else:
total_time =int((time.time() - start_time) *1000)
events.request_success.fire(
request_type='grpc',
name=name,
response_time=total_time,
response_length=0
)
return res
class GrpcUser(User):
abstract =True
def __init__(self, *args, **kwargs):
super(GrpcUser, self).__init__(*args, **kwargs)
self.client = GrpcClient()
然后下面是調(diào)用類
class BasicUser(GrpcUser):
"""
壓測(cè)任務(wù)類
執(zhí)行壓測(cè)命令： locust -f locust_file.py -u 10 -r 3 --headless
參數(shù)信息詳見locust官方網(wǎng)站：https://docs.locust.io/en/latest/running-locust-without-web-ui.html
"""
wait_time = between(2,
5) # locust自帶環(huán)境配置參數(shù)儿咱，詳見 https://docs.locust.io/en/latest/api.html?highlight=between#locust
# .wait_time.between
@task # task壓測(cè)任務(wù)語法糖，可以task(n)設(shè)置權(quán)重场晶，n越大混埠，任務(wù)權(quán)重越高
def test(self):
"""獲取未認(rèn)證賬號(hào)接口性能測(cè)試"""
name = "測(cè)試" # 任務(wù)名稱
env = "xxxx" # 服務(wù)名稱
path = "xxxx" # 服務(wù)path
parms = {
"person_user_ids": None, # 參數(shù)
"include_resign": None
}
self.client.connect(name=name, grpc_env=env, path=path, parms=parms) # 調(diào)用重寫后的connect方法，觸發(fā)壓測(cè)任務(wù)
是不是很簡(jiǎn)單 哈哈峰搪，加油~
