?????? 根据最近公司上线一个直播项目，但是有一天测试那边一脸苦逼的来找到我，问我有什么好工具可以实现高并发的测试，他使用jmeter限制了400个并发数（个人没去研究过不知道是否真的限制），后来我第一时间想到python写一个脚本，也没想多小就答应帮他搞一个测试脚本。
首先整理思：
- 要实现高并发不到两点
- 多进程
- 多线程
首先安装先要环境：
pip install websocketpip install threadpoolpip install websocket-clientpip install multiprocessing直接上脚本：
#!/usr/bin/python#-*- coding:utf-8 -*-#__author__ == ‘chenmingle‘import websocketimport timeimport threadingimport jsonimport multiprocessingfrom threadpool import ThreadPool, makeRequests#修改成自己的websocket地址WS_URL = "wss://ws.test.com/" #定义进程数processes=5#定义线程数（每个文件可能限制1024个，可以修改fs.file等参数）thread_num=1000def on_message(ws, message): ????print(message) ????passdef on_error(ws, error): ???print(error) ????passdef on_close(ws): ???print("### closed ###") ????passdef on_open(ws): ???def send_trhead(): ???????#设置你websocket的内容 ???????send_info = {"cmd": "refresh", "data": {"room_id": "58", "wx_user_id": 56431}} ???????#每隔10秒发送一下数据使链接不中断 ???????while True: ???????????time.sleep(10) ???????????ws.send(json.dumps(send_info)) ???t = threading.Thread(target=send_trhead) ???t.start()def on_start(num): ???time.sleep(num%20) ???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 = ThreadPool(thread_num) ???num = list() ???#设置开启线程的数量 ???for ir in range(thread_num): ???????num.append(ir) ???requests = makeRequests(on_start, num) ???[pool.putRequest(req) for req in requests] ???pool.wait()if __name__ == "__main__": ???#进程池 ???pool = multiprocessing.Pool(processes=processes) ???#设置开启进程的数量 ???for i in xrange(processes): ???????pool.apply_async(thread_web_socket) ???pool.close() ???pool.join()这样就完成一个高并发的websocket测试脚本，如果觉得这文章不错的，请在本人Github上点个star，感谢！！
参考资料：
- python thread pool websocket client
一个高并发的websocket脚本
原文地址：http://blog.51cto.com/legehappy/2144645