python实现websocket的客户端压力测试_Python

python实现websocket的客户端压力测试

2021-07-20 15:40刚刚登录 Python

这篇文章主要为大家详细介绍了python实现websocket的客户端压力测试，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下

使用python进行websocket的客户端压力测试，这个代码是从github上找到。然后简单修改了下。大神运用了进程池，以及线程池的内容。所以保存下来，学习学习

然后需要说明的是：本次用的python2.7，也尝试用python3.6，但是老实出现websocket-client包和python3不能兼容的情况，提示没有相关的方法。所以不得已最后又采用了python2

				?

									# -*- coding:utf-8 -*-

									# __author__ == 'chenmingle'

									import websocket

									import time

									import threading

									import json

									import multiprocessing

									import uuid

									from threadpool import ThreadPool, makeRequests

									# 修改成自己的websocket地址

									WS_URL = "xxxx"

									# 定义进程数

									processes = 4

									# 定义线程数（每个文件可能限制1024个，可以修改fs.file等参数）

									thread_num = 700

									index = 1

									def on_message(ws, message):

									 # print(message)

									 pass

									def on_error(ws, error):

									 print(error)

									 pass

									def on_close(ws):

									 # print("### closed ###")

									 pass

									def on_open(ws):

									 global index

									 index = index + 1

									 def send_thread():

									  # 设置你websocket的内容

									  # 每隔10秒发送一下数据使链接不中断

									  while True:

									   ws.send(u'hello服务器')

									   time.sleep(10)

									 t = threading.Thread(target=send_thread)

									 t.start()

									def on_start(num):

									 time.sleep(5)

									 # websocket.enableTrace(True)

									 ws = websocket.WebSocketApp(WS_URL + str(num),

									        on_message=on_message,

									        on_error=on_error,

									        on_close=on_close)

									 ws.on_open = on_open

									 ws.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 xrange(processes):

									  pool.apply_async(thread_web_socket)

									 pool.close()

									 pool.join()