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

更新时间：2026年03月06日 09:36:22 作者：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 支持丰富的配置选项，以下是常用参数说明：

参数	说明	默认值
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

参考资料

redis-py 官方文档

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

您可能感兴趣的文章:

一文了解发现并解决Redis热key与大key问题
热key是服务端的常见问题,本文主要介绍Redis热key与大key问题的解决方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
2024-05-05
redis实现主从模式(1主2从)
本文主要介绍了在Windows环境下搭建和测试Redis的主从复制模式,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
2024-12-12
Redis集群搭建(主从模式、哨兵模式、集群模式)
本文主要介绍了Redis集群搭建,主要包括主从模式、哨兵模式、集群模式这三种方式,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
2024-01-01
Redis Key的数量上限及优化策略分享
Redis 作为高性能的键值存储数据库,广泛应用于缓存、会话存储、排行榜等场景,但在实际使用中,开发者常常会关心一个问题：Redis 的 Key 数量是否有上限？本文将从 Redis Key 的理论上限出发,深入探讨 Redis Key 的管理策略,需要的朋友可以参考下
2025-03-03
深入探究RedisJSON模块的工作原理以及使用操作
Redis推出了RedisJSON模块,它允许开发者在Redis数据库中直接存储、查询和处理JSON数据,本文将详细介绍RedisJSON的工作原理、关键操作、性能优势以及使用场景,需要的朋友可以参考下
2024-05-05
Redis教程(九)：主从复制配置实例
这篇文章主要介绍了Redis教程(九)：主从复制配置实例,本文讲解了Redis的Replication、Replication的工作原理、如何配置Replication、应用示例等内容,需要的朋友可以参考下
2015-04-04
Redis中对大Key进行处理方式
这篇文章主要介绍了Redis中对大Key进行处理方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2025-06-06
Centos 7 如何安装Redis(推荐)
这篇文章主要介绍了Centos 7 如何安装Redis,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2020-12-12
Redis查看KEY的数据类型的方法和步骤
在Redis中,可以使用 TYPE 命令来查看指定key的数据类型,该命令会返回存储在指定key中的值的数据类型,本文给大家介绍了具体的使用方法和步骤,感兴趣的朋友可以参考下
2024-04-04
Linux、Windows下Redis的安装即Redis的基本使用详解
Redis是一个基于内存的key-value结构数据库，Redis 是互联网技术领域使用最为广泛的存储中间件，这篇文章主要介绍了Linux、Windows下Redis的安装即Redis的基本使用详解,需要的朋友可以参考下
2022-09-09