使用 Tornado websockets 和 RethinkDB 非阻塞地获取数据

Question

当我收到来自客户端的消息时，我想运行并行查询多个 RethinkDB 并立即将结果发送给客户端。

拦截方式如下。计数可能需要几分钟。我希望其他 return 更快的查询不会被计数查询阻止。

self.write_message({'count': r.db('public').table(message['table']).count().run(conn)})
self.write_message({'rows': r.db('public').table(message['table']).limit(10).run(conn)})

我怀疑我需要https://rethinkdb.com/blog/async-drivers/ and http://www.tornadoweb.org/en/stable/guide/async.html

的组合

我在想也许答案是让这两行类似于：

ioloop.IOLoop.current().add_callback(run_query, r.db('public').table(message['table']).count(), 'count', self)
ioloop.IOLoop.current().add_callback(run_query, r.db('public').table(message['table']).limit(10), 'rows', self)

我的运行-查询将是：

@gen.coroutine
def run_query(query, key, ws):
    conn = yield r.connect(host="localhost", port=28015)
    results = yield query.run(conn)
    ws.write_message({key: results})

Answer 1

tornado.gen doc 揭示解决方案：

You can also yield a list or dict of Futures, which will be started at the same time and run in parallel; a list or dict of results will be returned when they are all finished.

# do not forget about this
r.set_loop_type("tornado")

@gen.coroutine
def run_parallel(query, key, ws):
    conn = yield r.connect(host="localhost", port=28015)
    ret = yield {
        'count': r.db('public').table(message['table']).count().run(conn),
        'rows': r.db('public').table(message['table']).limit(10).run(conn)
    }
    ws.write_message(ret)

Yielding list 或 dict 直接具有重要的行为 - 如果任何 Futures 失败，yield 将立即 return 并且无论其他 Futures 是否完成都会 re-raise 异常。要绕过它，您可以改用 Mulit or multi_future.

注意：我真的不知道RethinkDB是否需要单独的连接，但我想展示一下概念。

使用 Tornado websockets 和 RethinkDB 非阻塞地获取数据

Nonblocking fetching of data with Tornado websockets and RethinkDB

python

tornado

rethinkdb

rethinkdb-python