這里主要是google的grpc接口進行壓測的一個栗子。
Locust是以HTTP為主要目標構建的。
但是，通過編寫鈎子觸發器request_success和 request_failure事件的自定義客戶端，可以輕松擴展到任何基於請求/響應的系統的負載測試 。
我們知道locust默認內部只封裝httplocust；使用的是requests中的session進行了封裝；
那么問題來了，如果我想測試其它協議怎么辦，比如websocket , grpc等等。
以grpc為例：
在開始之前，需要了解的幾點：
1>self.client: locust協議入口實例，我們只要重寫一個實例給client即可。
2>以下兩個鈎子事件，用來收集報告信息，否則寫好后執行你會發現收集不到性能數據
events.request_failure.fire()
events.request_success.fire()
具體步驟：
1>編寫GrpcClient類（主要用來替換掉self.client的http實例）
import time import grpc from grpctest.snowid import snowid_pb2,snowid_pb2_grpc from locust import (TaskSet,task,events,Locust) from gevent._semaphore import Semaphore all_locusts_spawned = Semaphore() all_locusts_spawned.acquire() def on_hatch_complete(**kwargs): all_locusts_spawned.release() events.hatch_complete += on_hatch_complete class GrpcClient(object): """重寫self.client""" def __init__(self): self.ht = '172.17.31.220' self.pt = '50073' def connect(self): """grpc實例"""try: #記錄開始時間 start_time = int(time.time()) #創建鏈接 self.conn = grpc.insecure_channel(self.ht +':'+self.pt) self.cl = snowid_pb2_grpc.snowidStub(channel=self.conn) #參數實例 args = snowid_pb2.GenerateSnowidRequest() args.uniqId = 10000 #此參數現在未起作用，可以為任意數字 #調用 res = self.cl.generateSnowid(args) total_time = int((time.time() - start_time) * 1000) if res.errCode != 0: raise Exception events.request_success.fire( request_type='grpc', name=r'/generateSnowid', response_time=total_time, response_length=0 ) except Exception as e: total_time = int((time.time() - start_time) * 1000) events.request_failure.fire( request_type='grpc', name='/generateSnowid', response_time=total_time, exception=e ) return res
注意：該類中定義了，grpc的常用調用操作；最主要是events.request_failure.fire和events.request_success.fire這兩個用來收集性能數據，如果不寫
報告收集不到性能數據。
上邊代碼只是以grpc舉例，其它協議也一樣，只要注意收集性能數據就可以。當然也可以不寫在這里。這個看代碼怎么寫了。
2>重寫一個HttpLocust類，我們這里叫做GrpcLocust類
class GrpcLocust(Locust): def __init__(self, *args, **kwargs): super(GrpcLocust, self).__init__(*args, **kwargs) self.client = GrpcClient()
注意：GrpcLocust從Locust繼承； 這里主要是將self.client重新實例成，我們第一部寫好的GrpcClient實例。
這樣一來我們通過self.client.xxxx就可以使用其方法
3>編寫TaskSet類
class GrpcTask(TaskSet): """壓測任務""" def on_start():
all_locusts_spawned.wait()
@task def generateSnowid(self): #grpc接口響應數據 res = self.client.connect() # print('errCode:{}'.format(res.errCode)) # print('result:{}'.format(res.result)) # print('errMsg:{}'.format(res.errMsg))
注意：
此類就是任務類，跟http的寫法一樣，只是這里用的self.client.xxxx已經變成了我們自已重寫的Grpc類，將原來的requests http替換了。
另外在TaskSet類中可以定義def on_start()：方法來定義，執行壓測任務，最先執行的方法。這個是重寫taskset中的on_start方法。
4>編寫站點類
class WebsiteUser(GrpcLocust): task_set = GrpcTask min_wait = 5000 max_wait = 9000
注意：站點類從第二步中的locust繼承
到這里代碼編寫完成，直接到cmd命令行執行
locust -f supperdiancan.py --no-web -c 10 -r 3 --run-time 10s
參數：
-f locust_file.py文件(locust任務文件)
-c 指定要生成的Locust用戶數
-r 每秒生成的用戶數
-n 用於0.8用於-n指定請求數
--run-time 或-t 指定測試的運行時間
注意：以上是以非web執行，當然你也可以用web執行。
以下是結果，從上可以看到，已經收集到了，請求數據。
Name # reqs # fails Avg Min Max | Median req/s -------------------------------------------------------------------------------------------------------------------------------------------- grpc /generateSnowid 22559 0(0.00%) 484 0 1007 | 480 673.60 -------------------------------------------------------------------------------------------------------------------------------------------- Total 22559 0(0.00%) 673.60 Percentage of the requests completed within given times Name # reqs 50% 66% 75% 80% 90% 95% 98% 99% 100% -------------------------------------------------------------------------------------------------------------------------------------------- grpc /generateSnowid 22559 480 640 740 790 890 950 980 990 1000 -------------------------------------------------------------------------------------------------------------------------------------------- Total 22559 480 640 740 790 890 950 980 990 1000