Python连接Redis连接池的具体使用_Python

什么是连接池

连接池是一种管理数据库连接的技术，它预先创建一定数量的连接并放入池中，当应用程序需要与数据库交互时，从池中获取一个连接使用，使用完毕后归还池中而非关闭。这种方式避免了频繁创建和销毁连接的开销，显著提升性能。

为什么需要连接池

在高频场景下（如限流、缓存），每次请求都创建新连接会造成巨大开销。连接池的核心优势包括：连接复用减少网络延迟、限制最大连接数防止资源耗尽、自动管理连接生命周期降低运维成本。

根据 redis-py 官方文档，高并发场景下使用连接池相比单连接可提升数倍性能。

同步连接池

基本用法

同步场景使用 redis.connectionpool 类，通过 from_url 方法可以从 redis url 创建连接池：

import redis

# 从 url 创建连接池
pool = redis.connectionpool.from_url(
    "redis://localhost:6379/0",
    max_connections=50,
    decode_responses=true,
)

# 创建客户端，复用连接池
client = redis.redis(connection_pool=pool)

# 使用
client.set("key", "value")
print(client.get("key"))

配置选项详解

connectionpool 支持丰富的配置选项，以下是常用参数说明：

参数	说明	默认值
host	redis 服务器地址	localhost
port	redis 服务器端口	6379
db	数据库编号	0
max_connections	最大连接数	2^31
socket_timeout	socket 超时时间（秒）	none
socket_connect_timeout	连接超时时间（秒）	none
socket_keepalive	是否保持连接活跃	true
health_check_interval	健康检查间隔（秒）	30
decode_responses	是否自动解码响应为字符串	false

完整配置示例：

pool = redis.connectionpool(
    host="localhost",
    port=6379,
    db=0,
    max_connections=50,           # 最大连接数
    socket_timeout=5.0,           # 操作超时 5 秒
    socket_connect_timeout=5.0,   # 连接超时 5 秒
    socket_keepalive=true,        # tcp keep-alive
    health_check_interval=30,     # 每 30 秒健康检查
    decode_responses=true,        # 返回字符串而非 bytes
)

多客户端共享连接池

多个 redis 客户端可以共享同一个连接池，连接会被复用：

pool = redis.connectionpool.from_url("redis://localhost:6379/0")

# 多个客户端共享连接池
r1 = redis.redis(connection_pool=pool)
r2 = redis.redis(connection_pool=pool)

# 数据共享
r1.set("shared_key", "value_from_r1")
print(r2.get("shared_key"))  # 输出: value_from_r1

# 关闭时需要关闭整个连接池
r1.close()
r2.close()
pool.disconnect()

阻塞连接池

blockingconnectionpool 在连接池耗尽时会阻塞等待，适用于需要严格控制并发的场景：

import redis

blocking_pool = redis.blockingconnectionpool(
    host="localhost",
    port=6379,
    max_connections=10,    # 最多 10 个连接
    timeout=20,            # 等待最多 20 秒
)
client = redis.redis(connection_pool=blocking_pool)

异步连接池

基本用法

异步场景使用 redis.asyncio 模块，连接池同样使用 connectionpool 类：

import asyncio
import redis.asyncio as aioredis

async def main():
    # 创建异步连接池
    pool = aioredis.connectionpool.from_url(
        "redis://localhost:6379/0",
        max_connections=20,
        decode_responses=true,
    )

    # 从连接池创建客户端
    client = aioredis.redis(connection_pool=pool)

    try:
        await client.set("async_key", "async_value")
        value = await client.get("async_key")
        print(f"value: {value}")
    finally:
        # 关闭客户端和连接池
        await client.aclose()
        await pool.disconnect()

asyncio.run(main())

使用from_pool方法

从连接池创建客户端的另一种方式是使用 from_pool 方法：

import redis.asyncio as redis

pool = redis.connectionpool.from_url("redis://localhost:6379/0")
client = redis.redis.from_pool(pool)

# 使用完毕后关闭
await client.aclose()

并发操作示例

异步连接池支持高并发场景下的批量操作：

import asyncio
import redis.asyncio as aioredis

async def batch_operations():
    pool = aioredis.connectionpool.from_url(
        "redis://localhost:6379/0",
        max_connections=100,
    )
    client = aioredis.redis(connection_pool=pool)

    try:
        # 批量设置
        tasks = [
            client.set(f"key:{i}", f"value:{i}")
            for i in range(100)
        ]
        await asyncio.gather(*tasks)

        # 批量读取
        get_tasks = [
            client.get(f"key:{i}")
            for i in range(100)
        ]
        results = await asyncio.gather(*get_tasks)
        print(f"读取到 {len(results)} 个值")
    finally:
        await client.aclose()
        await pool.disconnect()

asyncio.run(batch_operations())

高级配置

从 url 解析配置

使用 url 方式配置简洁且标准化，格式为 redis://host:port/db?options：

# 基础 url
pool = redis.connectionpool.from_url("redis://localhost:6379/0")

# 带密码
pool = redis.connectionpool.from_url("redis://:password@localhost:6379/0")

# 带额外参数
pool = redis.connectionpool.from_url(
    "redis://localhost:6379/0",
    max_connections=50,
    socket_timeout=5,
)

客户端类自定义

可以通过 redis 类的 connection_pool 参数指定连接池，也可以在子类中封装：

import redis
from redis.connection import connectionpool


class redisclient(redis.redis):
    """带连接池的 redis 客户端封装"""

    _pool: connectionpool | none = none

    def __new__(cls):
        if cls._pool is none:
            cls._pool = connectionpool.from_url(
                "redis://localhost:6379/0",
                max_connections=50,
                decode_responses=true,
            )
        return super().__new__(cls, connection_pool=cls._pool)


# 使用
client = redisclient()
client.set("key", "value")

单例模式实现

在固定配置的应用场景下（如限流），单例模式配合连接池是最佳实践：

import redis
from redis.connection import connectionpool


class redisclient(redis.redis):
    """redis 单例客户端（带连接池）"""

    _instance: "redisclient | none" = none
    _pool: connectionpool | none = none

    def __new__(cls):
        if cls._instance is none:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self) -> none:
        if redisclient._pool is none:
            redisclient._pool = connectionpool.from_url(
                "redis://localhost:6379/0",
                max_connections=50,
                decode_responses=true,
            )
            super().__init__(connection_pool=redisclient._pool)


# 全局单例
redis_client = redisclient()

# 使用
redis_client.set("key", "value")

最佳实践

1. 合理设置连接池大小

连接池大小应根据应用并发量和 redis 服务器性能设置。过小会导致连接等待，过大则浪费资源。一般建议设为服务并发数的 1.5 到 2 倍。

2. 正确管理生命周期

确保在应用退出时正确关闭连接池，避免资源泄漏：

# 同步
pool.disconnect()

# 异步
await client.aclose()
await pool.disconnect()

3. 配置适当的超时时间

设置 socket_timeout 防止长时间阻塞，设置 health_check_interval 保持连接健康：

pool = redis.connectionpool.from_url(
    "redis://localhost:6379/0",
    socket_timeout=5.0,
    socket_connect_timeout=5.0,
    health_check_interval=30,
)

常见问题

问题一：连接池耗尽

当 max_connections 设置过小或连接未正确归还时会出现此错误。解决方案包括增大连接池大小、检查连接是否正确释放、使用 blockingconnectionpool 并设置超时。

问题二：连接被关闭

redis 服务器默认配置下，空闲连接可能在一定时间后被关闭。通过设置 socket_keepalive=true 和适当的 health_check_interval 可以保持连接活跃。

问题三：异步连接池在多模块间共享

确保在程序结束时统一关闭连接池，避免部分模块关闭导致其他模块无法使用：

# 统一管理连接池
class poolmanager:
    _pool: connectionpool | none = none

    @classmethod
    def get_pool(cls):
        if cls._pool is none:
            cls._pool = connectionpool.from_url("redis://localhost:6379/0")
        return cls._pool

    @classmethod
    async def close(cls):
        if cls._pool:
            await cls._pool.disconnect()
            cls._pool = none