1

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

阻塞方式如下。计数可能需要几分钟。我希望其他可以更快返回的查询不会被计数查询所阻碍。

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/和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})
4

1 回答 1

1

tornado.gen doc揭示了解决方案：

您还可以生成 Futures 的列表或字典，它们将同时启动并并行运行；全部完成后将返回结果列表或字典。

# 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将立即返回并重新引发异常，无论其他 Futures 是否完成。要绕过它，您可以改用Mulit或multi_future。

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

于 2015-12-29T20:04:51.230 回答