python实现websocket的客户端压力测试
使用python进行websocket的客户端压力测试，这个代码是从github上找到。然后简单修改了下。大神运用了进程池，以及线程池的内容。所以保存下来，学习学习
然后需要说明的是：本次用的python2.7，也尝试用python3.6，但是老实出现websocket-client包和python3不能兼容的情况，提示没有相关的方法。所以不得已最后又采用了python2
#-*-coding:utf-8-*-
#__author__=='chenmingle'
importwebsocket
importtime
importthreading
importjson
importmultiprocessing
importuuid
fromthreadpoolimportThreadPool,makeRequests
#修改成自己的websocket地址
WS_URL="xxxx"
#定义进程数
processes=4
#定义线程数（每个文件可能限制1024个，可以修改fs.file等参数）
thread_num=700
index=1
defon_message(ws,message):
#print(message)
pass
defon_error(ws,error):
print(error)
pass
defon_close(ws):
#print("###closed###")
pass
defon_open(ws):
globalindex
index=index+1
defsend_thread():
#设置你websocket的内容
#每隔10秒发送一下数据使链接不中断
whileTrue:
ws.send(u'hello服务器')
time.sleep(10)
t=threading.Thread(target=send_thread)
t.start()
defon_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()
defthread_web_socket():
#线程池
pool_list=ThreadPool(thread_num)
num=list()
#设置开启线程的数量
foririnrange(thread_num):
num.append(ir)
requests=makeRequests(on_start,num)
[pool_list.putRequest(req)forreqinrequests]
pool_list.wait()
if__name__=="__main__":
#进程池
pool=multiprocessing.Pool(processes=processes)
#设置开启进程的数量
foriinxrange(processes):
pool.apply_async(thread_web_socket)
pool.close()
pool.join()
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。