4. aioredis 库入门：安装、连接池配置、基本命令的异步调用

在 Python 异步生态中，aioredis 是连接 Redis 的标准异步驱动。本章将围绕其安装、连接池配置以及核心命令的异步调用展开，为后续的缓存任务编排打下基础。

4.1 安装与版本选择

aioredis 目前有两个主要分支：aioredis 1.x（基于 asyncio）和 aioredis 2.x（基于 asyncio 并整合了 redis-py 的异步接口）。推荐使用 aioredis 2.x，它提供了更现代的 API 和更好的性能。

版本 安装命令 特点 推荐场景
aioredis 1.x pip install aioredis==1.3.1 独立驱动，API 较旧 遗留项目维护
aioredis 2.x pip install aioredis 与 redis-py 接口对齐，支持连接池、哨兵、集群 新项目首选
版本提示： 本课程所有示例均基于 aioredis 2.x。安装后可通过 pip show aioredis 确认版本。

4.2 连接池配置

连接池是异步 Redis 客户端的核心组件，它复用 TCP 连接，避免每次操作都创建新连接的开销。aioredis 2.x 使用 redis.asyncio.ConnectionPool 来管理连接。

4.2.1 基础连接池创建

import asyncio
from redis.asyncio import ConnectionPool, Redis

async def create_pool():
    # 创建连接池：最大连接数 10，最小空闲连接 2
    pool = ConnectionPool(
        host='localhost',
        port=6379,
        db=0,
        password=None,
        max_connections=10,
        min_idle_connections=2,
        socket_connect_timeout=5,
        socket_timeout=5,
        decode_responses=True  # 自动将 bytes 解码为 str
    )
    return pool

pool = asyncio.run(create_pool())

4.2.2 连接池参数详解

参数 类型 默认值 说明
host str 'localhost' Redis 服务器地址
port int 6379 Redis 服务器端口
db int 0 数据库编号
max_connections int None 连接池最大连接数，None 表示不限制
min_idle_connections int 0 保持的最小空闲连接数
socket_connect_timeout float None 连接超时（秒）
socket_timeout float None 读写超时（秒）
decode_responses bool False 是否自动解码响应为字符串
注意： 生产环境中务必设置 max_connections 和 socket_timeout，防止连接泄漏或无限等待。

4.3 基本命令的异步调用

aioredis 2.x 的 API 与同步 redis-py 高度一致，只需在调用前加上 await 即可。下面演示最常用的字符串、哈希、列表操作。

4.3.1 字符串操作

async def string_ops(redis: Redis):
    # SET / GET
    await redis.set('user:1001:name', 'Alice')
    name = await redis.get('user:1001:name')
    print(f'Name: {name}')  # 输出: Name: Alice

    # SETEX（带过期时间）
    await redis.setex('session:token:abc123', 3600, 'valid')
    ttl = await redis.ttl('session:token:abc123')
    print(f'TTL: {ttl}')  # 输出: TTL: 3600

    # MSET / MGET（批量操作）
    await redis.mset({'key1': 'val1', 'key2': 'val2'})
    vals = await redis.mget('key1', 'key2')
    print(f'Values: {vals}')  # 输出: Values: ['val1', 'val2']

4.3.2 哈希操作

async def hash_ops(redis: Redis):
    # HSET / HGET
    await redis.hset('user:1001', mapping={'name': 'Bob', 'age': 30})
    name = await redis.hget('user:1001', 'name')
    print(f'Name: {name}')  # 输出: Name: Bob

    # HGETALL
    user_data = await redis.hgetall('user:1001')
    print(f'User data: {user_data}')  # 输出: User data: {'name': 'Bob', 'age': '30'}

    # HEXISTS / HDEL
    exists = await redis.hexists('user:1001', 'email')
    print(f'Email exists: {exists}')  # 输出: Email exists: False

4.3.3 列表操作

async def list_ops(redis: Redis):
    # LPUSH / RPUSH
    await redis.lpush('queue:task', 'task1', 'task2')
    await redis.rpush('queue:task', 'task3')

    # LRANGE
    tasks = await redis.lrange('queue:task', 0, -1)
    print(f'Tasks: {tasks}')  # 输出: Tasks: ['task2', 'task1', 'task3']

    # LPOP / RPOP
    first = await redis.lpop('queue:task')
    print(f'Popped: {first}')  # 输出: Popped: task2

    # LLEN
    length = await redis.llen('queue:task')
    print(f'Length: {length}')  # 输出: Length: 2

4.4 连接池与命令调用的完整流程

下面的 SVG 图展示了从连接池创建到命令执行、连接归还的完整生命周期。

aioredis 连接池与命令调用流程 aioredis 连接池与命令调用流程 异步应用 ConnectionPool 空闲连接队列 [conn1, conn2, ...] max_connections=10 Redis Server 获取连接 发送命令 返回结果 返回数据 连接获取/归还 获取: 从空闲队列弹出 归还: 压入空闲队列 关键点： • 连接池负责连接的复用与生命周期管理，避免频繁创建/销毁 TCP 连接。 • 每次命令调用从池中获取连接，执行完毕后自动归还，无需手动释放。

4.5 完整示例：连接池 + 命令调用

将连接池创建与命令调用整合到一个异步函数中，展示完整的“获取连接 → 执行命令 → 归还连接”流程。

import asyncio
from redis.asyncio import ConnectionPool, Redis

async def main():
    # 1. 创建连接池
    pool = ConnectionPool(
        host='localhost',
        port=6379,
        db=0,
        max_connections=10,
        decode_responses=True
    )

    # 2. 从连接池获取 Redis 客户端实例
    redis = Redis(connection_pool=pool)

    try:
        # 3. 执行命令
        await redis.set('course:title', 'Redis Async Task Orchestration')
        title = await redis.get('course:title')
        print(f'Course title: {title}')

        # 4. 批量操作
        await redis.hset('course:info', mapping={
            'instructor': 'Expert',
            'duration': '30 chapters'
        })
        info = await redis.hgetall('course:info')
        print(f'Course info: {info}')

    finally:
        # 5. 关闭连接池（释放所有连接）
        await redis.close()
        await pool.disconnect()

if __name__ == '__main__':
    asyncio.run(main())
最佳实践： 在应用启动时创建一次连接池，全局复用；在应用关闭时统一释放。避免在每次请求中创建新的连接池。

4.6 连接池配置对比：单机 vs 哨兵 vs 集群

不同部署模式下，连接池的配置略有差异。下表总结了三种常见场景的配置要点。

部署模式 连接池类 关键参数 示例
单机 ConnectionPool host, port ConnectionPool(host='localhost', port=6379)
哨兵 SentinelConnectionPool service_name, sentinel_kwargs SentinelConnectionPool(service_name='mymaster', sentinel_kwargs={'password': 'xxx'})
集群 RedisCluster（内置连接池） startup_nodes, cluster_error_retry_attempts RedisCluster(startup_nodes=[{'host': '127.0.0.1', 'port': 7000}])
注意： 集群模式下不要手动创建 ConnectionPool，应使用 RedisCluster 自动管理连接池。

4.7 常见问题与避坑

  • 连接池耗尽： 如果 max_connections 设置过小，高并发时会出现 ConnectionError: max connections reached。建议根据并发量估算，一般设置为 CPU核心数 × 2 或更高。
  • 连接泄漏： 使用 Redis 实例后未调用 close()，会导致连接无法归还池中。务必使用 try...finally 或 async with 上下文管理器。
  • 响应解码： 设置 decode_responses=True 后，所有响应自动转为 str；如果后续需要 bytes，请单独处理。
  • 超时设置： 不设置 socket_timeout 可能导致命令永久阻塞。建议根据业务容忍度设置 3-10 秒。

4.8 本章小结

通过本章学习，你已掌握：

  • aioredis 2.x 的安装与版本选择
  • 连接池的创建、参数配置与生命周期管理
  • 字符串、哈希、列表等基本命令的异步调用
  • 连接池与命令调用的完整流程
  • 不同部署模式下的连接池配置差异
← 上一章 📖 返回目录 下一章 →