Python連接Redis連接池的具體使用

更新時(shí)間：2026年03月06日 09:36:22 作者：liguohhhhh

在Redis中,同步和異步場(chǎng)景分別使用redis.ConnectionPool和redis.asyncio.ConnectionPool,配置選項(xiàng)包括最大連接數(shù)、超時(shí)時(shí)間等,多客戶端和單例模式也是連接池的常見應(yīng)用,下面就來(lái)詳細(xì)的介紹一下

什么是連接池

連接池是一種管理數(shù)據(jù)庫(kù)連接的技術(shù)，它預(yù)先創(chuàng)建一定數(shù)量的連接并放入池中，當(dāng)應(yīng)用程序需要與數(shù)據(jù)庫(kù)交互時(shí)，從池中獲取一個(gè)連接使用，使用完畢后歸還池中而非關(guān)閉。這種方式避免了頻繁創(chuàng)建和銷毀連接的開銷，顯著提升性能。

為什么需要連接池

在高頻場(chǎng)景下（如限流、緩存），每次請(qǐng)求都創(chuàng)建新連接會(huì)造成巨大開銷。連接池的核心優(yōu)勢(shì)包括：連接復(fù)用減少網(wǎng)絡(luò)延遲、限制最大連接數(shù)防止資源耗盡、自動(dòng)管理連接生命周期降低運(yùn)維成本。

根據(jù) redis-py 官方文檔，高并發(fā)場(chǎng)景下使用連接池相比單連接可提升數(shù)倍性能。

同步連接池

基本用法

同步場(chǎng)景使用 redis.ConnectionPool 類，通過(guò) from_url 方法可以從 Redis URL 創(chuàng)建連接池：

import redis

# 從 URL 創(chuàng)建連接池
pool = redis.ConnectionPool.from_url(
    "redis://localhost:6379/0",
    max_connections=50,
    decode_responses=True,
)

# 創(chuàng)建客戶端，復(fù)用連接池
client = redis.Redis(connection_pool=pool)

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

配置選項(xiàng)詳解

ConnectionPool 支持豐富的配置選項(xiàng)，以下是常用參數(shù)說(shuō)明：

參數(shù)	說(shuō)明	默認(rèn)值
host	Redis 服務(wù)器地址	localhost
port	Redis 服務(wù)器端口	6379
db	數(shù)據(jù)庫(kù)編號(hào)	0
max_connections	最大連接數(shù)	2^31
socket_timeout	socket 超時(shí)時(shí)間（秒）	None
socket_connect_timeout	連接超時(shí)時(shí)間（秒）	None
socket_keepalive	是否保持連接活躍	True
health_check_interval	健康檢查間隔（秒）	30
decode_responses	是否自動(dòng)解碼響應(yīng)為字符串	False

完整配置示例：

pool = redis.ConnectionPool(
    host="localhost",
    port=6379,
    db=0,
    max_connections=50,           # 最大連接數(shù)
    socket_timeout=5.0,           # 操作超時(shí) 5 秒
    socket_connect_timeout=5.0,   # 連接超時(shí) 5 秒
    socket_keepalive=True,        # TCP Keep-Alive
    health_check_interval=30,     # 每 30 秒健康檢查
    decode_responses=True,        # 返回字符串而非 bytes
)

多客戶端共享連接池

多個(gè) Redis 客戶端可以共享同一個(gè)連接池，連接會(huì)被復(fù)用：

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

# 多個(gè)客戶端共享連接池
r1 = redis.Redis(connection_pool=pool)
r2 = redis.Redis(connection_pool=pool)

# 數(shù)據(jù)共享
r1.set("shared_key", "value_from_r1")
print(r2.get("shared_key"))  # 輸出: value_from_r1

# 關(guān)閉時(shí)需要關(guān)閉整個(gè)連接池
r1.close()
r2.close()
pool.disconnect()

阻塞連接池

BlockingConnectionPool 在連接池耗盡時(shí)會(huì)阻塞等待，適用于需要嚴(yán)格控制并發(fā)的場(chǎng)景：

import redis

blocking_pool = redis.BlockingConnectionPool(
    host="localhost",
    port=6379,
    max_connections=10,    # 最多 10 個(gè)連接
    timeout=20,            # 等待最多 20 秒
)
client = redis.Redis(connection_pool=blocking_pool)

異步連接池

基本用法

異步場(chǎng)景使用 redis.asyncio 模塊，連接池同樣使用 ConnectionPool 類：

import asyncio
import redis.asyncio as aioredis

async def main():
    # 創(chuàng)建異步連接池
    pool = aioredis.ConnectionPool.from_url(
        "redis://localhost:6379/0",
        max_connections=20,
        decode_responses=True,
    )

    # 從連接池創(chuàng)建客戶端
    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:
        # 關(guān)閉客戶端和連接池
        await client.aclose()
        await pool.disconnect()

asyncio.run(main())

使用from_pool方法

從連接池創(chuàng)建客戶端的另一種方式是使用 from_pool 方法：

import redis.asyncio as redis

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

# 使用完畢后關(guān)閉
await client.aclose()

并發(fā)操作示例

異步連接池支持高并發(fā)場(chǎng)景下的批量操作：

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:
        # 批量設(shè)置
        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)} 個(gè)值")
    finally:
        await client.aclose()
        await pool.disconnect()

asyncio.run(batch_operations())

高級(jí)配置

從 URL 解析配置

使用 URL 方式配置簡(jiǎn)潔且標(biāo)準(zhǔn)化，格式為 redis://host:port/db?options：

# 基礎(chǔ) URL
pool = redis.ConnectionPool.from_url("redis://localhost:6379/0")

# 帶密碼
pool = redis.ConnectionPool.from_url("redis://:password@localhost:6379/0")

# 帶額外參數(shù)
pool = redis.ConnectionPool.from_url(
    "redis://localhost:6379/0",
    max_connections=50,
    socket_timeout=5,
)

客戶端類自定義

可以通過(guò) Redis 類的 connection_pool 參數(shù)指定連接池，也可以在子類中封裝：

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")

單例模式實(shí)現(xiàn)

在固定配置的應(yīng)用場(chǎng)景下（如限流），單例模式配合連接池是最佳實(shí)踐：

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")

最佳實(shí)踐

1. 合理設(shè)置連接池大小

連接池大小應(yīng)根據(jù)應(yīng)用并發(fā)量和 Redis 服務(wù)器性能設(shè)置。過(guò)小會(huì)導(dǎo)致連接等待，過(guò)大則浪費(fèi)資源。一般建議設(shè)為服務(wù)并發(fā)數(shù)的 1.5 到 2 倍。

2. 正確管理生命周期

確保在應(yīng)用退出時(shí)正確關(guān)閉連接池，避免資源泄漏：

# 同步
pool.disconnect()

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

3. 配置適當(dāng)?shù)某瑫r(shí)時(shí)間

設(shè)置 socket_timeout 防止長(zhǎng)時(shí)間阻塞，設(shè)置 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,
)

常見問(wèn)題

問(wèn)題一：連接池耗盡

當(dāng) max_connections 設(shè)置過(guò)小或連接未正確歸還時(shí)會(huì)出現(xiàn)此錯(cuò)誤。解決方案包括增大連接池大小、檢查連接是否正確釋放、使用 BlockingConnectionPool 并設(shè)置超時(shí)。

問(wèn)題二：連接被關(guān)閉

Redis 服務(wù)器默認(rèn)配置下，空閑連接可能在一定時(shí)間后被關(guān)閉。通過(guò)設(shè)置 socket_keepalive=True 和適當(dāng)?shù)?health_check_interval 可以保持連接活躍。

問(wèn)題三：異步連接池在多模塊間共享

確保在程序結(jié)束時(shí)統(tǒng)一關(guān)閉連接池，避免部分模塊關(guān)閉導(dǎo)致其他模塊無(wú)法使用：

# 統(tǒng)一管理連接池
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