Redis

关注公众号 jb51net

关闭
首页 > 数据库 > Redis > Python Redis连接池

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

作者：liguohhhhh

在Redis中,同步和异步场景分别使用redis.ConnectionPool和redis.asyncio.ConnectionPool,配置选项包括最大连接数、超时时间等,多客户端和单例模式也是连接池的常见应用,下面就来详细的介绍一下

什么是连接池

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

为什么需要连接池

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

根据 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 支持丰富的配置选项，以下是常用参数说明：

参数说明默认值
hostRedis 服务器地址localhost
portRedis 服务器端口6379
db数据库编号0
max_connections最大连接数2^31
socket_timeoutsocket 超时时间（秒）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

参考资料

到此这篇关于Python连接Redis连接池的具体使用的文章就介绍到这了,更多相关Python Redis连接池内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:
阅读全文