python-websocket压力测试

devtools/2025/3/18 13:11:35/

一.websocket简介及安装

使用pip命令安装websocket库:pip install websocket-client

websocket.WebSocketApp 是对 websocket.WebSocket 的封装,支持自动定时发送 PING 帧,支持事件驱动方式的数据帧接收,可用于长期的 WebSocket 连接。

websocket中就有建立连接connect、发送消息send等函数可供使用,但是websocket.WebSocketApp将这些都封装好了,只用在实例化的时候传入自定义函数即可,更方便。

WebSocketApp也是websocket中的一个类。要使用WebSocketApp中的回调函数需要传入一系列的可调用对象。在实例化该类时传入构造函数中的on_open、on_message、on_error就需要传入一系列的可调用对象,例如自定义的函数。
 

二.方法

  运行WebSocketApp的事件循环,先创建webSocket对象,然后connect连接服务器,之后一直循环运行接收数据帧,回调对应函数处理数据帧;当websocket客户端被关闭后,将调用on_close()方法然后结束循环返回;当循环中发生异常时被捕捉,然后依次调用on_error(),on_close()方法,然后结束循环返回。

python"># -*- coding:utf-8 -*-import websocket
import time
import threading
import multiprocessing
from threadpool import ThreadPool, makeRequests"""
websocket长链接
"""# websocket地址
WS_URL = "ws://124.222.224.186:8800"
# 定义进程数
processes = 2
# 定义线程数(每个文件可能限制1024个,可以修改fs.file等参数)
thread_num = 3
# 拼接发送的消息
sendMsg = '{"appid":"futures","cover":0,"event":[\{"type":"exchange_rate","toggle":1,"expireTime":86400},\{"type":"accountInfo_USDT","toggle":1,"expireTime":'def on_message(ws, message):print("***message***", message)def on_error(ws, error):print("***error***", error)def on_close(ws, *args):print("***closed***")def on_open(ws):def send_thread():# 设置你websocket的内容# 每隔10秒发送一下数据使链接不中断while True:# ws.send('hello服务器' + str(time.time()))ws.send(sendMsg + str(time.time()) + '}]}')time.sleep(10)t = threading.Thread(target=send_thread)t.start()def on_start(num):time.sleep(5)ws = websocket.WebSocketApp(WS_URL,on_message=on_message,on_error=on_error,on_close=on_close)ws.on_open = on_openws.run_forever()def thread_web_socket():# 线程池pool_list = ThreadPool(thread_num)num = list()# 设置开启线程的数量for ir in range(thread_num):num.append(ir)requests = makeRequests(on_start, num)[pool_list.putRequest(req) for req in requests]pool_list.wait()if __name__ == "__main__":# 进程池pool = multiprocessing.Pool(processes=processes)# 设置开启进程的数量for i in range(processes):pool.apply_async(thread_web_socket)pool.close()pool.join()

2个进程运行,各运行3个线程,建立6个连接,每10秒发送一次数据:


http://www.ppmy.cn/devtools/168061.html

相关文章

TypeScript接口 interface 高级用法完全解析

TypeScript接口 interface 高级用法完全解析 mindmaproot(TypeScript接口高级应用)基础强化可选属性只读属性函数类型高级类型索引签名继承与合并泛型约束设计模式策略模式工厂模式适配器模式工程实践声明合并类型守卫装饰器集成一、接口核心机制深度解析 1.1 类型兼容性原理 …

Maven学习

Maven是用来构建项目,管理依赖的 Maven项目结构 main主要写实现代码,test是写测试代码不会被打包 resource放静态资源,webapp存放web开发内容。 pom文件包括项目版本号内容以及相关配置以及后续依赖 Maven依赖导入,就是直接写…

【通义千问】蓝耘智算 | 智启未来:蓝耘MaaS×通义QwQ-32B引领AI开发生产力

【作者主页】Francek Chen 【专栏介绍】 ⌈ ⌈ ⌈人工智能与大模型应用 ⌋ ⌋ ⌋ 人工智能(AI)通过算法模拟人类智能,利用机器学习、深度学习等技术驱动医疗、金融等领域的智能化。大模型是千亿参数的深度神经网络(如ChatGPT&…

音视频入门基础:RTCP专题(1)——RTCP官方文档下载

一、引言 实时传输控制协议(Real-time Transport Control Protocol或RTP Control Protocol或简写RTCP)是实时传输协议(RTP)的一个姐妹协议。RTCP由《RFC 3550》定义(取代废弃的《RFC 1889》)。RTP使用一个…

【AI 加持下的 Python 编程实战 2_04】第三章:GitHub Copilot 在 Python 函数设计中的正确打开方式(含本地实操)

【全新第二版《Learn AI-assisted Python Programming》封面】 写在前面 本篇是全书的第一处精华内容,从完全零基础小白的角度详细介绍了 Copilot 在 Python 函数设计中的用法,包括提示词的正确写法、基于 AI 工具的函数设计基本工作流程等,虽…

人工智能中神经网络是如何进行学习的

前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住分享一下给大家。点击跳转到网站。https://www.captainbed.cn/north 文章目录 引言神经网络的学习过程1. 前向传播2. 计算损失3. 反向传播反向传播的步骤 4. 参数更新5. 重…

python-leetcode 54.全排列

题目: 给定不含重复数字的数组nums,返回其所有可能的全排列,可以按任意顺序返回答案 回溯法 一种通过探索所有可能的候选解来找出所有的解的算法。如果候选解被确认不是一个解(或者至少不是最后一个解),回溯算法会通…

基于 YOLOv8 和 PyQt5 的火焰、烟雾检测

目标检测是计算机视觉领域的一个重要研究方向,广泛应用于安防监控、自动驾驶、工业检测等领域。近年来,随着深度学习技术的快速发展,YOLO(You Only Look Once)系列算法因其速度快、精度高而备受关注。本文将介绍如何利用 YOLOv8 和 PyQt5 开发一个视频目标检测应用,实现从…