locust壓測websocket協議

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

但是，通過編寫觸發器request_success和 request_failure事件的自定義客戶端，可以輕松擴展到任何基於請求/響應的系統的負載測試 。

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

那么問題來了，如果我想測試其它協議怎么辦，比如websocket  , grpc等等。

 

以websoket為例：

 

在開始之前，需要了解的幾點：

1>self.client: locust協議入口實例，我們只要重寫一個實例給client即可。

2>以下兩個事件，用來收集報告信息，否則寫好后執行你會發現收集不到性能數據

events.request_failure.fire()

events.request_success.fire()

 

具體步驟：

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

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