`
from tornado import ioloop, web
import json
import aioredis
settings = {
"debug" : True,
"redis":"redis://:@127.0.0.1:6379/0",
}
class Home(web.RequestHandler):
def prepare(self):
self.json_argument = {}
# if self.request.method.lower() in ("get", "post", "put", "delete", "patch"):
# if self.request.headers.get("Content-Type").lower() == "application/json":
# self.json_argument = json.loads(self.request.body)
async def get(self):# 2.0之后，aioredis 提供了 from_url 方法，直接从 url 中创建连接# redis = await aioredis.create_connection(self.application.settings["redis"])# redis = await aioredis.create_connection(self.application.settings["redis"])# await redis.execute("setex", "name", 60, "chen",)# data = await redis.get("name")# print(data.decode())# self.write(data.decode())redis = aioredis.from_url(self.application.settings["redis"])async with redis.client() as conn:await conn.setex("name", 60, "chen")data = await conn.get("name")print(data.decode())self.write(data.decode())
async def redis_pool():
# Redis client bound to pool of connections (auto-reconnecting).
redis = aioredis.from_url(
"redis://localhost", encoding="utf-8", decode_responses=True
)
await redis.set("my-key", "value")
val = await redis.get("my-key")
print(val)
def make_app():
return web.Application(handlers=[
(r"/", Home)],
**settings,
)
if name == 'main':
app = make_app()
app.listen(8888)
ioloop.IOLoop.current().start()
`