locust 压测 websocket

* 安装 python 3.8

https://www.python.org/

py --version

* 安装 locust

pip install locust==2.5.1 -i http://pypi.douban.com/simple/

pip install locust==2.5.1 -i https://pypi.mirrors.ustc.edu.cn/simple/

locust -V

备注：-i 是切换下载源

* 安装依赖

pip install websockets

pip install websocket

pip install websocket_client

* 新建文件 main.py

import time
import websocket
from locust import User, task, between, events

class WebSocketUser(User):
    wait_time = between(1, 2)

    def on_start(self):
        self.connection_status = False
        try:
            self.client = websocket.create_connection("ws://127.0.0.1:7777")
            self.connection_status = True
            events.request_success.fire(request_type="ws connect", name="connect Success",  response_time=0, response_length=0)
        except Exception as e:
            print(e)
            events.request_failure.fire(request_type="ws connect", name="connect Fail", exception=e, response_time=0, response_length=0)
        

    def on_stop(self):
        if self.connection_status:
            self.client.close()

    @task
    def ws_task(self):
        if self.connection_status:
            try:
                request = '{"cmd":"2","subcmd":"1","data":{"account":"daming","password":"daming"}}'
                start_time = time.time()
                self.client.send(request)
                response = self.client.recv()
                # print("response=", response)
                total_time = int((time.time() - start_time) * 1000)
                events.request_success.fire(request_type="ws send", name="Send Success", 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="ws send", name="Send Fail", exception=e, response_time=total_time, response_length=0)

* 进入到路径下

locust -f main.py