locust壓測websocket協議

Locust是以HTTP爲主要目標構建的。html

可是,經過編寫觸發器request_success和 request_failure事件的自定義客戶端,能夠輕鬆擴展到任何基於請求/響應的系統的負載測試 。web

咱們知道locust默認內部只封裝httplocust;使用的是requests中的session進行了封裝;api

那麼問題來了,若是我想測試其它協議怎麼辦,好比websocket  , grpc等等。websocket

 

以websoket爲例:session

 

在開始以前,須要瞭解的幾點:socket

1>self.client: locust協議入口實例,咱們只要重寫一個實例給client便可。分佈式

2>如下兩個事件,用來收集報告信息,不然寫好後執行你會發現收集不到性能數據性能

events.request_failure.fire()
events.request_success.fire()

 

具體步驟:測試

1>編寫WebSocketClient類(主要用來替換掉self.client的http實例)url

class WebSocketClient(object):

    def __init__(self, host):
        self.host = host
        self.ws = websocket.WebSocket()

    def connect(self, burl):
        start_time = time.time()
        try:
            self.conn = self.ws.connect(url=burl)
        except websocket.WebSocketTimeoutException as e:
            total_time = int((time.time() - start_time) * 1000)
            events.request_failure.fire(request_type="websockt", name='urlweb', response_time=total_time, exception=e)
        else:
            total_time = int((time.time() - start_time) * 1000)
            events.request_success.fire(request_type="websockt", name='urlweb', response_time=total_time, response_length=0)
        return self.conn

    def recv(self):
        return self.ws.recv()

    def send(self, msg):
        self.ws.send(msg)

注意:該類中定義了,websocket的經常使用操做,連接、接收、發送;最主要是events.request_failure.fire和events.request_success.fire這兩個用來收集性能數據,若是不寫

報告收集不到性能數據。

上邊代碼只是以websocket舉例,其它協議也同樣,只要注意收集性能數據就能夠。固然也能夠不寫在這裏。這個看代碼怎麼寫了。

 

2>重寫一個HttpLocust類,咱們這裏叫作WebsoketLoscust類

class WebsocketLocust(Locust):
    def __init__(self, *args, **kwargs):
        super(WebsocketLocust, self).__init__(*args, **kwargs)
        self.client = WebSocketClient(self.host)

注意:WebsocketLocust從Locust繼承; 這裏主要是將self.client從新實例成,咱們第一部寫好的websocketClient實例。 

這樣一來咱們經過self.client.xxxx就可使用其方法

3>編寫TaskSet類

class SupperDianCan(TaskSet):

    @task
    def test_baidu(self):
        self.url = 'wss://xxxxxx.xxxx.com/cart/chat?sid=11303&table_no=103&user_id=ofZjWs40HxEzvV08l6m4PnqGbxqc_2_1_&version=2'

        self.data = {}

        self.client.connect(self.url)
        while True:
            recv = self.client.recv()
            print(recv)
            if eval(recv)['type'] == 'keepalive':
                self.client.send(recv)
            else:
                self.client.send(self.data)

注意:此類就是任務類,跟http的寫法同樣,只是這裏用的self.client.xxxx已經變成了咱們自已重寫的websocket類,將原來的requests http替換了。

4>編寫站點類

class WebsiteUser(WebsocketLocust):

    task_set = SupperDianCan

    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執行。

如下是結果,從上能夠看到,已經收集到了,請求數據。

Percentage of the requests completed within given times
 Name                                                           # reqs    50%    66%    75%    80%    90%    95%    98%    99%   100%
--------------------------------------------------------------------------------------------------------------------------------------------
 websockt urlweb                                                    10     48     50     59     65    540    540    540    540    540
--------------------------------------------------------------------------------------------------------------------------------------------
 Total                                                              10     48     50     59     65    540    540    540    540    540

 

以上並不是是在分佈式下運行,如要分佈式,請參考其它。

官方參考文檔,請穿越:https://docs.locust.io/en/stable/testing-other-systems.html

相關文章
相關標籤/搜索