在异步编程日益普及的今天，如果你还在用同步的redis-py客户端访问Redis，那么每一个Redis请求都会阻塞整个事件循环，让asyncio的并发优势荡然无存。aioredis正是为解决这个问题而生的库，它完全基于Python的asyncio框架实现，所有Redis操作都以协程方式执行。本文将从安装配置、连接池管理、常见命令操作到集群模式，系统地讲解aioredis的使用方法。
aioredis简介与安装
aioredis是一个纯异步的Redis客户端，底层完全基于asyncio实现。它的API设计与redis-py保持高度一致，熟悉redis-py的开发者几乎可以无缝迁移。需要注意的是，从2.0版本开始，aioredis直接整合了redis-py的代码，API风格发生了较大变化，同时也提供了原生的集群支持。如果你使用的是1.x版本，建议尽快迁移到2.x，因为1.x已经停止维护。
安装非常简单，通过pip即可完成：
pip install aioredis
如果你使用的是Python 3.8及以上版本，aioredis 2.x可以直接使用async/await语法。安装完成后，可以通过以下代码验证版本：
import aioredis print(aioredis.__version__)
创建连接与连接池的使用
在任何异步操作之前，首先需要建立与Redis服务器的连接。aioredis 2.x中最推荐的方式是使用Redis对象配合连接池，而不是每次请求都新建连接。频繁创建连接会带来TCP三次握手和AUTH认证的开销，在高并发下性能损失明显。
下面是一个创建连接池的标准写法：
import asyncio
import aioredis
async def main():
# 创建Redis连接，内部自动维护连接池
redis = aioredis.from_url(
"redis://127.0.0.1:6379",
encoding="utf-8",
decode_responses=True,
max_connections=50
)
# 写入数据
await redis.set("name", "async_user")
# 读取数据
value = await redis.get("name")
print(value)
# 关闭连接池
await redis.close()
asyncio.run(main())这里有几个关键参数值得注意：decode_responses=True会让返回值自动从bytes解码为str，省去手动decode的麻烦；max_connections限制了连接池的最大连接数，防止连接无限增长；encoding指定编码格式，一般保持utf-8即可。
在Web框架中，推荐在应用启动时创建一次Redis连接池，在处理请求时复用。以FastAPI为例，可以在lifespan中管理连接的生命周期：
from contextlib import asynccontextmanager
from fastapi import FastAPI
import aioredis
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.redis = aioredis.from_url("redis://127.0.0.1:6379")
yield
await app.state.redis.close()
app = FastAPI(lifespan=lifespan)
@app.get("/count")
async def get_count():
count = await app.state.redis.incr("visit_count")
return {"count": count}这种模式保证了整个应用生命周期内只存在一个连接池，所有请求共享，是生产环境的标准做法。切记不要在每个请求处理函数中单独创建Redis实例，那样会导致连接数失控，最终耗尽服务器的文件描述符。
常见命令操作与管道批量执行
aioredis支持Redis的全部命令，方法名与Redis命令一一对应。字符串、哈希、列表、集合、有序集合等数据结构的操作都是await调用。看几个典型例子：
import asyncio
import aioredis
async def demo(redis):
# 字符串操作，支持过期时间
await redis.set("token", "abc123", ex=3600)
# 哈希操作
await redis.hset("user:1", mapping={"name": "张三", "age": "28"})
user = await redis.hgetall("user:1")
# 列表操作
await redis.lpush("queue", "task1", "task2")
tasks = await redis.lrange("queue", 0, -1)
# 带过期判断的key操作
ttl = await redis.ttl("token")
print(ttl)当需要一次性执行大量命令时，逐条await会带来多次网络往返。Redis的pipeline（管道）机制可以把多条命令打包后一次性发送，大幅减少网络延迟。aioredis中通过pipeline()方法使用：
async def pipe_demo(redis):
pipe = redis.pipeline()
pipe.set("k1", "v1")
pipe.set("k2", "v2")
pipe.incr("counter")
# 一次性执行，返回所有结果
results = await pipe.execute()
print(results) # [True, True, 1]管道默认不保证原子性，如果需要严格的事务语义，可以使用pipe.transaction()或者MULTI/EXEC命令组合。管道对于批量写入场景的优化效果非常显著，实测在批量写入1000个key时，使用管道可以将耗时从数秒降低到几十毫秒级别。
连接Redis Cluster与异常处理
当数据量增大到单机Redis无法承载时，就需要使用Redis Cluster。aioredis 2.x提供了内置的集群支持，使用aioredis.RedisCluster即可：
import asyncio
import aioredis
async def cluster_demo():
redis = aioredis.RedisCluster(
host="127.0.0.1",
port=7000,
decode_responses=True
)
await redis.set("cluster_key", "hello")
print(await redis.get("cluster_key"))
await redis.close()
asyncio.run(cluster_demo())集群模式下客户端会自动根据CRC16算法计算key所属的槽位，并把命令路由到正确的节点，开发者无需关心分片细节。需要注意的是，涉及多个key的命令（如MSET）要求所有key分布在同一个槽位，否则会报CROSSSLOT错误，可以通过hash tag语法（如user:{1000}:name）强制相关key落在同一槽。
关于异常处理，网络抖动和Redis重启是生产环境必然会遇到的问题。aioredis在连接断开时会抛出aioredis.ConnectionError，建议对关键操作添加重试逻辑：
import asyncio
import aioredis
async def set_with_retry(redis, key, value, retries=3):
for i in range(retries):
try:
return await redis.set(key, value)
except (aioredis.ConnectionError, aioredis.TimeoutError):
if i == retries - 1:
raise
await asyncio.sleep(0.5 * (i + 1))最后补充一点迁移建议：由于aioredis 2.x本质上是redis-py的异步分支，官方更推荐直接使用redis.asyncio模块（即from redis import asyncio as aioredis），API完全一致且维护更活跃。理解了aioredis的用法之后，切换到官方异步客户端也只需修改一行import语句，学习成本几乎为零。
aioredisRedis异步连接Python异步编程修改时间：2026-09-09 04:14:36