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

Python  /  管理员 发布于 7年前   310

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

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

# -*- coding:utf-8 -*-# __author__ == 'chenmingle' import websocketimport timeimport threadingimport jsonimport multiprocessingimport uuidfrom threadpool import ThreadPool, makeRequests # 修改成自己的websocket地址WS_URL = "xxxx"# 定义进程数processes = 4# 定义线程数（每个文件可能限制1024个，可以修改fs.file等参数）thread_num = 700index = 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()

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。