python----/ws/发送.ipynb

In [ ]:

import websocket
import json

def on_open(ws):
    print("WebSocket 连接已建立")
    # 发送文本消息
    ws.send("Hello, WebSocket Server!")
    # 发送 JSON 消息
    data = {"type": "message", "content": "这是JSON数据"}
    ws.send(json.dumps(data))

def on_message(ws, message):
    print(f"收到服务器消息: {message}")

def on_close(ws, close_status_code, close_msg):
    print("连接已关闭")

if __name__ == "__main__":
    # WebSocket 服务器地址
    ws_url = "ws://192.168.0.125:58080/webSocket"  # 测试用公共服务器
    # 创建 WebSocket 连接
    ws = websocket.WebSocketApp(
        ws_url,
        on_open=on_open,
        on_message=on_message,
        on_close=on_close
    )
    # 启动连接（阻塞模式）
    ws.run_forever()

---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
Cell In[1], line 22
     17         # # 接收服务器响应
     18         # response = await ws.recv()
     19         # print(f"收到服务器消息: {response}")
     21 if __name__ == "__main__":
---> 22     asyncio.get_event_loop().run_until_complete(send_message())

File d:\Anaconda3\Lib\asyncio\base_events.py:663, in BaseEventLoop.run_until_complete(self, future)
    652 """Run until the Future is done.
    653 
    654 If the argument is a coroutine, it is wrapped in a Task.
   (...)
    660 Return the Future's result, or raise its exception.
    661 """
    662 self._check_closed()
--> 663 self._check_running()
    665 new_task = not futures.isfuture(future)
    666 future = tasks.ensure_future(future, loop=self)

File d:\Anaconda3\Lib\asyncio\base_events.py:622, in BaseEventLoop._check_running(self)
    620 def _check_running(self):
    621     if self.is_running():
--> 622         raise RuntimeError('This event loop is already running')
    623     if events._get_running_loop() is not None:
    624         raise RuntimeError(
    625             'Cannot run the event loop while another loop is running')

RuntimeError: This event loop is already running