利用python對websocket進行並發壓測

簡述

產品經理鑒於運營反饋並對程序的websocket長連接保持懷疑的態度，讓我對websocket服務器進行壓力測試，我內心是拒絕的。

 

開發思路

查閱websocket的相關資料，查到python的websocket有兩個連接方式，長連接：WebSocketApp，短連接：create_connection。

使用 WebSocketApp 的話，我沒辦法獲取websocket服務器端返回的數據，這個我還在研究，這里使用 create_connection 來進行壓測。

連接：

ws = create_connection("websocket的服務器地址")

獲取連接狀態：

print("獲取連接狀態：", ws.connected)　　# True為連接成功，False為連接失敗

發送消息至服務器並接收返回結果：

initialize_data = '{"code":2001, "msg": "","data": ""}'
ws.send(initialize_data)
result_initialize = ws.recv()        # 獲取返回結果
print("接收結果：", result_initialize)

因為create_connection為短連接，所以這一系列下來后連接會斷開，所以這個時候需要一個東西來維持連接，長連接叫心跳，那短連接這里我願成為：心臟起搏器(while True)

完整代碼：

import threading
from websocket import create_connection
import time

thread = []
def socket():
    while True:
        ws = create_connection("websocket服務器地址")
        # 2、獲取連接狀態
        print("獲取連接狀態：", ws.connected)
        print("獲取服務器返回的連接結果", ws.recv())
        if ws.connected == True:
            initialize = '{"code":2001, "msg": "","data": {}}'
            ws.send(initialize)
            result_initialize = ws.recv()  # 獲取返回結果
            print("接收結果：", result_initialize)
if __name__ == "__main__":
    for i in range(25):
        t = threading.Thread(target=socket, args=())
        t.start()