Tornado WebSocket跨独立线程向客户端发消息无响应问题求助

你的代码不生效是三个核心错误导致的：

• 变量作用域错误：你在类方法、实例方法中引用了cls.waiters/self.waiters，但WSHandler类本身没有定义这个类属性，你定义的waiters是模块级全局变量，遍历的时候拿不到正确的客户端集合，自然发不出消息。
• IOLoop线程绑定错误：Tornado的IOLoop是严格和启动它的线程绑定的，你在独立子线程中调用IOLoop.instance()拿到的大概率是子线程自己生成的未启动的IOLoop实例，往这个实例上加的回调永远不会被执行，这就是你回调里的打印都没输出的核心原因。
• 跨线程操作违规：Tornado 5+版本禁止在非IOLoop线程直接调用write_message，对连接集合的增删、遍历如果不在IOLoop线程执行，还会出现竞态问题。

首先修正作用域错误，然后显式传递主线程正在运行的IOLoop实例给工作线程，所有连接操作、消息发送逻辑全部放到IOLoop线程执行。

修正后的服务端代码 #

import tornado.web
import tornado.websocket
import tornado.ioloop
import json
import threading

class MainHandler(tornado.web.RequestHandler):
    def get(self):
        print("[HTTP](MainHandler) User Connected.")
        self.render("index.html")

class WSHandler(tornado.websocket.WebSocketHandler):
    waiters = set()  # 把连接集合定义为类属性，避免全局变量污染
    cache = []
    cache_size = 200

    def open(self):
        print(list(self.__class__.waiters))
        print('[WS] Connection was opened.')
        self.__class__.waiters.add(self)
        print(list(self.__class__.waiters))

    def on_message(self, message):
        print('[WS] Incoming message:', message)
        if message == "toggle":
            for client in self.__class__.waiters:
                print("fired")

    def on_close(self):
        print(list(self.__class__.waiters))
        print('[WS] Connection was closed.')
        self.__class__.waiters.remove(self)
        print(list(self.__class__.waiters))

# 消息发送方法，必须在IOLoop线程执行
def send_msg_to_all_clients(JSONmsg):
    print("recv msg to send")
    print(JSONmsg)
    for client in WSHandler.waiters:
        print('sent to client')
        client.write_message(json.dumps(JSONmsg))

# 示例工作线程函数
def your_worker_func(loop):
    # 线程内业务逻辑
    import time
    while True:
        time.sleep(3)
        msg_json={"type":"StatusUpdate","message":"Connected","message1":"123412","Message2":"Calculating RSSI","Message3":""}
        # 把发送逻辑丢到主线程IOLoop执行
        loop.add_callback(send_msg_to_all_clients, msg_json)

if __name__ == "__main__":
    app = tornado.web.Application([
        (r"/", MainHandler),
        (r"/ws", WSHandler),
    ])
    app.listen(8888)
    # 获取主线程正在运行的IOLoop实例，显式传给工作线程
    main_io_loop = tornado.ioloop.IOLoop.current()
    # 启动工作线程，传入主线程IOLoop实例
    worker_thread = threading.Thread(target=your_worker_func, args=(main_io_loop,), daemon=True)
    worker_thread.start()
    main_io_loop.start()

独立线程调用注意点 #

不要在工作线程内部调用IOLoop.instance()/IOLoop.current()获取事件循环，必须从主线程把已经启动的main_io_loop实例传进工作线程，再通过add_callback调度发送逻辑。

这个方案本质是在服务内部启动一个本地WebSocket客户端，和服务端建立回环连接，工作线程把要推送的消息发给这个内部客户端，再由客户端通过WS连接发给服务端，服务端收到后广播给所有真实用户。

• 优点：不需要跨线程调度IOLoop，适配旧版本的线程模型
• 缺点：多了一层本地WS转发，性能比直接调度差，需要额外加内部鉴权逻辑避免外部伪造消息
  实现逻辑：

1. 服务端新增内部消息识别逻辑，收到带特定标识的消息时，广播给所有普通客户端
2. 工作线程启动时，用tornado.websocket.websocket_connect建立到本地WS服务的连接，所有要推送的消息直接通过这个连接发送即可

• 所有对WSHandler.waiters集合的增删、遍历操作，必须全部放在IOLoop线程执行，不要在其他线程直接操作这个集合，否则会出现迭代时集合被修改、连接对象状态不一致等问题。
• 如果需要给指定客户端发消息，可以给每个连接实例加用户标识属性，遍历的时候匹配标识再发送即可。
• Tornado 6+版本移除了旧版IOLoop.instance()的全局默认实例逻辑，必须显式传递主线程IOLoop实例给其他线程，不要依赖全局方法获取。

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