python-3.x - Tornado + aioredis : why are my redis calls blocking?

Question

我尝试在 Tornado 和 Redis 上构建一个具有两个 API 端点的简单系统:

1. API 从 Redis 读取一个值，或者等待这个值存在(使用 BRPOP : value = yield from redis.brpop("test"))<
2. 写入此值的 API(使用 LPUSH :redis.lpush("test", "the value"))。

所以我希望能够以任何顺序调用这些 API。事实上，如果我调用 2. then 1.，它会按预期工作，对 1. 的调用会立即返回值。

问题是，如果我先调用 1. 然后调用 2.，那么两个请求都会被阻止而永远不会返回。

同时，在请求阻塞时，我仍然可以直接在 Redis 中使用 LPUSH/BRPOP，即使是同一个键。同样，我可以在 Tornado 中调用其他 Handlers。所以我猜这个 block 既不在 Redis 也不在 Tornado 中，而是在我对 aioredis 的使用中？也许异步循环？但我不明白我错在哪里。有什么建议吗？

感谢您的帮助。

这是我的代码:

import tornado.ioloop
import tornado.web
from tornado import web, gen
from tornado.options import options, define
import aioredis
import asyncio


class WaitValueHandler(tornado.web.RequestHandler):
    @asyncio.coroutine
    def get(self):
        redis = self.application.redis
        value = yield from redis.brpop("test")
        self.write("I received a value: %s" % value)


class WriteValueHandler(tornado.web.RequestHandler):
    @asyncio.coroutine
    def get(self):
        redis = self.application.redis
        res = yield from redis.lpush("test", "here is the value")
        self.write("Ok ")


class Application(tornado.web.Application):
    def __init__(self):
        tornado.ioloop.IOLoop.configure('tornado.platform.asyncio.AsyncIOMainLoop')

        handlers = [
            (r"/get", WaitValueHandler),
            (r"/put", WriteValueHandler)
        ]

        super().__init__(handlers, debug=True)

    def init_with_loop(self, loop):
        self.redis = loop.run_until_complete(
            aioredis.create_redis(('localhost', 6379), loop=loop)
        )

if __name__ == "__main__":
    application = Application()
    application.listen(8888)

    loop = asyncio.get_event_loop()
    application.init_with_loop(loop)
    loop.run_forever()

Answer 1

好的，我知道为什么了，as the doc states :

Blocking operations (like blpop, brpop or long-running LUA scripts) in shared mode mode will block connection and thus may lead to whole program malfunction.

This blocking issue can be easily solved by using exclusive connection for such operations:

redis = await aioredis.create_redis_pool(
    ('localhost', 6379),
    minsize=1,
    maxsize=1)

async def task():
   # Exclusive mode
   with await redis as r:
       await r.set('key', 'val')
asyncio.ensure_future(task())
asyncio.ensure_future(task())
# Both tasks will first acquire connection.