Tornado/WebSocket新手求助：实现桌面客户端WebSocket交互及等待响应功能

嘿，作为Tornado和WebSocket领域的新手，你的需求其实很有代表性，我来一步步帮你拆解实现思路和代码示例，帮你搞定这几个功能！

一、基础WebSocket客户端搭建 #

首先我们需要定义一个继承自tornado.websocket.WebSocketHandler的客户端类，这是Tornado处理WebSocket通信的核心：

import tornado.websocket
import tornado.ioloop
import uuid
from tornado.concurrent import Future

class MyWebSocketClient(tornado.websocket.WebSocketHandler):
    # 存储客户端实例，方便外部调用
    _instance = None
    # 用于等待响应的消息ID-Future映射
    _pending_futures = {}

    @classmethod
    def get_instance(cls):
        if not cls._instance:
            # 这里可以根据实际情况初始化，比如连接到服务器地址
            cls._instance = cls()
        return cls._instance

    def open(self):
        print("WebSocket连接已建立")

    def on_close(self):
        print("WebSocket连接已关闭")
        # 清理pending的futures
        for future in self._pending_futures.values():
            future.set_exception(Exception("连接已关闭"))
        self._pending_futures.clear()
        self._instance = None

二、需求1：接收服务器数据并通知外部组件 #

在on_message方法里处理服务器下发的数据，完成校验后，通过回调函数或者信号通知外部组件：

def initialize(self, data_callback=None):
        # 初始化时传入外部组件的回调函数
        self.data_callback = data_callback

    def on_message(self, message):
        try:
            # 1. 解析服务器发来的消息（假设是JSON格式）
            import json
            server_data = json.loads(message)
            
            # 2. 执行数据校验逻辑（这里替换成你的实际校验规则）
            if self._validate_data(server_data):
                # 3. 校验通过，向外部组件发送信号/调用回调
                if self.data_callback:
                    self.data_callback(server_data)
            else:
                print("数据校验失败")
        except Exception as e:
            print(f"处理服务器消息出错: {e}")

    def _validate_data(self, data):
        # 示例校验逻辑：检查是否包含required字段
        return "required_field" in data

使用的时候，外部组件可以这样注册回调：

def external_component_signal(data):
    # 外部组件收到数据后的处理逻辑
    print(f"外部组件收到有效数据: {data}")

# 初始化客户端并传入回调
client = MyWebSocketClient.get_instance()
client.initialize(data_callback=external_component_signal)

三、需求2：向服务器发送无需响应的数据（已实现） #

确认一下你的发送方法是不是类似这样，通常直接调用write_message即可：

def send_no_response_data(self, data):
        # 把数据转成JSON字符串发送（根据服务器格式调整）
        import json
        self.write_message(json.dumps(data))

四、需求3：外部函数向服务器发送数据 #

关键是让外部函数能拿到WebSocket客户端的实例，我们已经在类里实现了单例模式get_instance()，所以外部函数可以这样调用：

def request_masks_export(self, export_dir):
    wrapped_command = {
        "clientAction": "export_masks",
        "exportDir": export_dir
    }
    # 获取客户端实例并发送数据
    client = MyWebSocketClient.get_instance()
    # 注意：如果外部函数在非Tornado IO线程（比如GUI主线程），需要用add_callback
    tornado.ioloop.IOLoop.current().add_callback(
        lambda: client.send_no_response_data(wrapped_command)
    )

五、新增需求：发送消息并等待响应 #

因为Tornado是异步框架，我们用Future来实现异步等待响应的功能：

async def send_and_wait_response(self, message_data):
        # 生成唯一消息ID，用于匹配响应
        message_id = str(uuid.uuid4())
        message_data["messageId"] = message_id
        
        # 创建Future对象，用于等待响应
        future = Future()
        self._pending_futures[message_id] = future
        
        # 发送消息
        self.write_message(json.dumps(message_data))
        
        try:
            # 等待响应，超时时间可以自定义
            response = await tornado.gen.with_timeout(tornado.ioloop.IOLoop.current().time() + 10, future)
            return response
        except tornado.gen.TimeoutError:
            del self._pending_futures[message_id]
            raise Exception("等待响应超时")
        except Exception as e:
            del self._pending_futures[message_id]
            raise e

    def on_message(self, message):
        try:
            import json
            server_data = json.loads(message)
            
            # 先检查是否是响应消息（包含messageId）
            if "messageId" in server_data and server_data["messageId"] in self._pending_futures:
                future = self._pending_futures.pop(server_data["messageId"])
                future.set_result(server_data)
                return
            
            # 处理普通数据（原需求1的逻辑）
            if self._validate_data(server_data):
                if self.data_callback:
                    self.data_callback(server_data)
            else:
                print("数据校验失败")
        except Exception as e:
            print(f"处理服务器消息出错: {e}")

然后外部异步函数可以这样调用：

async def example_wait_response():
    client = MyWebSocketClient.get_instance()
    try:
        response = await client.send_and_wait_response({
            "clientAction": "get_config"
        })
        print(f"收到服务器响应: {response}")
    except Exception as e:
        print(f"请求失败: {e}")

# 在Tornado IO循环中运行异步函数
tornado.ioloop.IOLoop.current().add_callback(example_wait_response)

六、桌面应用的事件循环整合 #

如果你的桌面应用用的是PyQt/Tkinter等GUI框架，需要把Tornado的IO循环和GUI的事件循环整合，避免阻塞。比如用Tkinter的话，可以这样做：

import tkinter as tk

def run_tornado_loop():
    tornado.ioloop.IOLoop.current().add_callback(lambda: None)
    tornado.ioloop.IOLoop.current().start()

root = tk.Tk()
# 启动Tornado IO循环在后台线程
import threading
threading.Thread(target=run_tornado_loop, daemon=True).start()

# 运行GUI主循环
root.mainloop()

内容的提问来源于stack exchange，提问作者randomCoder1