当前位置：首页 > news >正文

python --连接websocket

news 2025/8/3 2:18:37

如果只是模拟js端发送接收的话，已经有了websocket server的话，只有client就好了

pip install websocket-client

#-*- encoding:utf-8 -*-import sys
sys.path.append("..")
from socket import *
import json, time, threading
from websocket import create_connectionclass Client():def __init__(self):#调用create_connection方法，建立一个websocket链接,链接是自己的链接self.ws = create_connection("ws://127.0.0.1:10010/xxxx")#建一个线程，监听服务器发送给客户端的数据self.trecv = threading.Thread(target=self.recv)self.trecv.start()#发送方法，聊天输入语句时调用，此处默认为群聊ALLdef send(self,content):#这里的msg要根据实际需要自己写msg={"type":"POST","content":content}msg = json.dumps(msg)self.ws.send(msg)#接收服务端发送给客户的数据，只要ws处于连接状态，则一直接收数据def recv(self):try:while self.ws.connected:result = self.ws.recv()print "received msg:"+str(result)except Exception,e:pass if __name__ == '__main__':c= Client()#建立链接后，就可以按照需要自己send了c.send(content)

二

接下来，你可以使用以下代码来连接WebSocket服务器：

import websocketdef 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()