反应产生了龙卷风

Question

我想通过websocket创建信使。我的逻辑是：User_1通过龙卷风处理程序向User_2发送消息（json），在龙卷风服务器上检查消息（def send_message_to_RDB_parallel）（一些请求到RDB、PostgreSQL），然后User_1收到响应，User_2收到消息。

检查RDB（def send_message_to_RDB_parallel）的请求-可能会阻塞我的龙卷风服务器。正因为如此，我想通过Celery（使用RabbitMQ）完成它，或者只是放弃了它。据我所知，它可以帮助我解锁龙卷风服务器。但我需要在完成时得到回复。我可以用芹菜也可以不用，但我得不到回应。。当我破坏我的龙卷风服务器（按下Ctrl-C）时，我会看到一个错误，比如"…对象不可调用"

如何获取响应并发送（self.write_message()）？

在这个例子中，我试图只使用收益

class MessagesHandler(tornado.websocket.WebSocketHandler):
    ...
    def on_message(self, mess):
    ...
        self.send_message_to_RDB(thread_id=thread_id,
                                     sender_id=self.user_id,
                                     recipient_id=recipient_id,
                                     message=message['msg'],
                                     time=datetime.datetime.now(datetime.timezone.utc),
                                     check=True)
    ...                              
    @tornado.gen.coroutine
    def send_message_to_RDB(self, thread_id, sender_id, recipient_id, message, time, check):
        response = yield tornado.gen.Task(send_message_to_RDB_parallel(thread_id=thread_id,
                                             sender_id=sender_id,
                                             recipient_id=recipient_id,
                                             message=message,
                                             time=time,
                                             check=check))
        if response.result[0] is False:
            self.write_message(response.result[1])

def send_message_to_RDB_parallel(thread_id, sender_id, recipient_id, message, time, check=False):
    """
    Send message to rdb. Check thread. One recipient_id !
    """
  #  tf__ = False
    if check is True:
        if recipient_id == sender_id:
            return False, to_json_error(MessengerRecipientEqualSenderServerMessage[1])
        if User.objects.filter(id=recipient_id,
                               is_deleted=False,
                               is_active=True,
                               is_blocked=False).exists() is False:
            return False, to_json_error("Wrong User")
    ...    
    else:
        me = Message()
        me.text = message
        me.thread_id = thread_id
        me.sender_id = sender_id
        me.datetime = time
        me.save()
    return True, None

Answer 1

有几个常见错误：

1. send_message_to_RDB_parallel不是异步的，甚至没有回调参数，但您试图将其与gen.Task一起使用-不会设置任何结果
2. on_message是一个协程，它在send_message_to_RDB中被调用，坚果它不屈服（等待）
3. gen.Task接受一个函数（以及可选的附加参数）并运行它，但它实际上是您调用它的代码，而不是传递

因为2）不会引发任何进一步的错误，这样您就可以在^C之后看到它们。必须阅读http://www.tornadoweb.org/en/stable/guide/async.html

解决方案

当然，您可以使用celeni并异步等待结果（Tornado-celeni集成技巧）。。

但如果您使用Postgres，我建议您使用现有的异步库（使用SQLAlchemy和Tornado保存API输出异步）：

• momoko-基于Tornado的postgres客户端，它不是ORM
• aiopg-基于postgres异步的客户端（Tornado 4.3及以上版本），支持sqlalchemy查询生成器