locust socektio协议压测
2024-09-08 18:44:48
# -*-coding:UTF-8 -*-
from locust import HttpLocust, TaskSet, task, TaskSequence, Locust, events
import time, socketio, random class SocketClient(object): def __init__(self, host):
self.host = host
self.ws = socketio.Client()
self.connected = False def connect(self, burl):
start_time = time.time()
# 判断是否连接,连接就不执行
if not self.connected:
try:
self.conn = self.ws.connect(url=burl)
self.connected = True
except socketio.client.exceptions as e:
total_time = int((time.time() - start_time) * 1000) events.request_failure.fire(request_type="socket", name='建立连接', response_time=total_time, exception=e) else:
total_time = int((time.time() - start_time) * 1000)
events.request_success.fire(request_type="socket", name='建立连接', response_time=total_time,
response_length=0) return self.conn def emit(self, event, data, namespace):
start_time = time.time()
try:
self.ws.emit(event, data, namespace)
except Exception as e:
total_time = int((time.time() - start_time) * 1000) events.request_failure.fire(request_type="socket", name='加入/发送文本', response_time=total_time, exception=e) else:
total_time = int((time.time() - start_time) * 1000)
events.request_success.fire(request_type="socket", name='加入/发送文本', response_time=total_time,
response_length=0) def wait(self):
return self.ws.wait() class socketLocust(Locust):
def __init__(self, *args, **kwargs):
super(socketLocust, self).__init__(*args, **kwargs)
self.client = SocketClient(self.host) class SupperDianCan(TaskSet):
@task
def test(self):
self.url = "https://xxx.com:6302"
self.client.connect(self.url)
#
self.client.emit('join-room', {
"user_id": 123,
"team_id": 345
}, namespace='/mz-chat')
#
self.client.emit('new_message', {
"user_id": 123,
"team_id": 345,
"content": "我来了"}, namespace='/mz-chat')
# period = int(time.strftime("%S", time.localtime())) + 9
# while 1:
# nowtime = time.strftime("%S", time.localtime())
# self.s_sleep(5)
# self.client.emit('new_message', {
# "user_id": 123,
# "team_id": 345,
# "content": "data"}, namespace='/mz-chat')
# if int(nowtime) > period:
# break
# self.wait() class dc_socket(socketLocust):
task_set = SupperDianCan
min_wait = 1000
max_wait = 2000 if __name__ == "__main__":
import os os.system("locust -f locust1.py --no-web -c1 -r1 ")
最新文章
- WPF +MVVM(Caliburn.Micro)项目框架
- PHP常量、变量作用域详解(一)
- Python_转义字符
- Python之路【第十一篇续】前端初识之CSS
- CSS选择器实现搜索功能 驱动过滤搜索技术
- hdu 2594 Simpsons’ Hidden Talents KMP
- 下载文件的一种简单方法js
- css first-letter实现首字(字母)下沉效果
- 在vs2005中使用SVN进行版本管理
- JS调用android逻辑方法
- windows下配置Nginx+Mysql+Php7
- HDU4497GCD and LMC最大公约数与最小公倍数
- ubuntu 常用命令集
- parse_str 与 http_build_query的使用
- SQL 经典语句
- Java序列化和反序列化,你该知道得更多
- symfony command
- (二 -5) 天猫精灵接入Home Assistant-自动发现Mqtt设备--电风扇
- PHP7 网络编程(六)Socket和IO多路复用【待】
- 创建一个Django项目的基本步骤