在Python中，可以使用`websocket`库来连接WebSocket服务器。以下是两种实现Python连接WebSocket的示例代码： ### 示例一 ```python import websocket def on_message(ws, message): print("Received message:", message) def on_error(ws, error): print("Error:", error) def on_close(ws): print("WebSocket closed") def on_open(ws): print("WebSocket connected") # 连接的WebSocket服务器地址 websocket_url = "ws://example.com/websocket" # 创建WebSocket连接 websocket.enableTrace(True) # 可选，用于调试。打开调试信息输出 ws = websocket.WebSocketApp(websocket_url, on_message=on_message, on_error=on_error, on_close=on_close) ws.on_open = on_open # 开始WebSocket连接 ws.run_forever() ``` 此示例中，定义了几个回调函数`on_message`、`on_error`、`on_close`和`on_open`，分别用于处理接收到的消息、错误、连接关闭和连接建立事件。创建`WebSocketApp`对象并传入服务器地址和回调函数，最后调用`run_forever`方法来保持连接并处理事件 [^2]。 ### 示例二 ```python import _thread import json import ssl import time import websocket def on_message(ws, message): print(ws) print(message) def on_error(ws, error): print(ws) print(error) def on_close(ws): print(ws) print("closed") def on_open(ws): def run(*args): # 建立连接后自动订阅 ws.send(json.dumps({})) while True: time.sleep(10) ws.send(json.dumps({"ping": int(time.time() * 1000)})) _thread.start_new_thread(run, ()) if __name__ == "__main__": url = "" websocket.enableTrace(True) ws = websocket.WebSocketApp(url, on_open=on_open, on_message=on_message, on_error=on_error, on_close=on_close) # wss跳过SSL验证 ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) ``` 该示例同样定义了回调函数，在`on_open`函数中创建了一个新线程，用于在连接建立后自动发送订阅消息，并每隔10秒发送一个`ping`消息。最后使用`run_forever`方法保持连接，同时通过`sslopt`参数跳过SSL验证 [^3]。