python - 当 on_message Tornado .websocket 出现阻塞时我该怎么办？

Question

我是 Tornado 的新手。我正在尝试用 Tornado 构建一个聊天服务器代理，我从网络客户端收到消息，通常它只需要将其发送回来，但是，我需要将这些消息发送给另一个服务器优先，问题来了，等待其他服务器响应需要花费很多时间，我需要使其无阻塞，但是当我使用 Tornado 的匿名方法时，它根本不起作用，求助我，非常感谢你!

这是我的伪代码的一部分:

class ClientWSConnectienter(websocket.WebSocketHandler):

_thread_pool = ThreadPoolExecutor(20)

def initialize(self, room_handler):
    #chat room initiate
    self.__rh = room_handler

@run_on_executor(executor='_thread_pool')
def worker(self,msg):
    #send the msg to another server
    pmessage=send_msg_to_server(msg) 
    return pmessage

@tornado.web.asynchronous
@tornado.gen.coroutine
def on_message(self, message):
    #this will blocking for too much time,and I want make it no-blocking
    pmessage=yeild worker(msg) 
    #send the recive pmessage to others client
    room.write_message(pmessage) 
    self.finish()

显然，它不起作用，我得到了这样的东西:

error:websocket cannot use this method

那么，我该怎么办？非常感谢

但是在我重新编辑代码后，它仍然阻塞在任务部分。我不知道为什么，这仍然是我代码的一部分重新编辑:

class ClientWSConnection(websocket.WebSocketHandler):

def initialize(self, room_handler):
    self.queue = tornado.queues.Queue()

def open(self, client_id):
    IOLoop.current().spawn_callback(self.loop)

def on_message(self, message):
    self.queue.put(msg)

def on_close(self):
    self.queue.put(None)

@coroutine
def loop(self):
    while 1:
        msg=yield self.queue.get()
        if msg is None:
            return
        msg=yield self.worker(msg)
        pmessage = msg
        room.write_message(pmessage)
@coroutine
def worker(self,msg):
    #need to send the other server,blocking here
    time.sleep(10)
    raise Return(msg)

Answer 1

我认为错误消息来自您对 finish() 的调用，这对于 websockets 没有意义(您的意思是 close() 吗？)。 (此外，没有必要同时使用 @asynchronous 和 @coroutine；单独使用 @coroutine 就足够了)

但是还有一个更大的问题:请记住，当重写父类(super class)中定义的方法时，只有在文档允许的情况下才能将它们设为协程(因为协程的调用方式与常规方法不同)。 WebSocketHandler.on_message 目前(从 Tornado 4.3 开始)不支持协程。

因此您需要使用队列将其交给另一个任务。像这样的事情:

class MyHandler(WebSocketHandler):
    def initialize(self):
        self.queue = tornado.queues.Queue()

    def on_open(self):
        IOLoop.current().spawn_callback(self.loop)

    def one_message(self, msg):
        self.queue.put(msg)

    def on_connection_close(self):
        self.queue.put(None)

    @coroutine
    def loop(self):
        while True:
            msg = yield self.queue.get()
            if msg is None:
                return
            pmessage = yield self.worker(msg)
            self.write_message(pmessage)