最新接手一个新项目, 在跑测试的时候, 有大量的报错 Connection closed by server

  File "/usr/local/lib/python3.12/site-packages/redis/asyncio/client.py", line 616, in execute_command
    return await conn.retry.call_with_retry(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/redis/asyncio/retry.py", line 62, in call_with_retry
    await fail(error)
  File "/usr/local/lib/python3.12/site-packages/redis/asyncio/client.py", line 603, in _disconnect_raise
    raise error
  File "/usr/local/lib/python3.12/site-packages/redis/asyncio/retry.py", line 59, in call_with_retry
    return await do()
           ^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/redis/asyncio/client.py", line 590, in _send_command_parse_response
    return await self.parse_response(conn, command_name, **options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/redis/asyncio/client.py", line 637, in parse_response
    response = await connection.read_response()
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/redis/asyncio/connection.py", line 543, in read_response
    response = await self._parser.read_response(
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/redis/_parsers/resp2.py", line 82, in read_response
    response = await self._read_response(disable_decoding=disable_decoding)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/redis/_parsers/resp2.py", line 90, in _read_response
    raw = await self._readline()
          ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/redis/_parsers/base.py", line 221, in _readline
    raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
redis.exceptions.ConnectionError: Connection closed by server.

根据经验判断, 是 redis 连接的问题. 稍微看了下代码实现

项目实现

class RedisPool:
    def __init__(self):
        self.async_redis_pool = redis.ConnectionPool(...)

    def client(self) -> redis.Redis:
        return redis.Redis.from_pool(self.async_redis_pool)

pool = RedisPool()

with pool.client() as rds:
	...

try:
    rds = pool.client()
    ...
except:
    pass
finally:
    rds.aclose()

Redis 库实现

class Redis(
    AbstractRedis, AsyncRedisModuleCommands, AsyncCoreCommands, AsyncSentinelCommands
):
	async def aclose(self, close_connection_pool: Optional[bool] = None) -> None:
	    """
	    Closes Redis client connection
	
	    :param close_connection_pool: decides whether to close the connection pool used
	    by this Redis client, overriding Redis.auto_close_connection_pool. By default,
	    let Redis.auto_close_connection_pool decide whether to close the connection
	    pool.
	    """
	    conn = self.connection
	    if conn:
	        self.connection = None
	        await self.connection_pool.release(conn)
	    if close_connection_pool or (
	        close_connection_pool is None and self.auto_close_connection_pool
	    ):
	        await self.connection_pool.disconnect()

class ConnectionPool:
    async def release(self, connection: AbstractConnection):
        """Releases the connection back to the pool"""
        # Connections should always be returned to the correct pool,
        # not doing so is an error that will cause an exception here.
        self._in_use_connections.remove(connection)
        self._available_connections.append(connection)

分析

根据 redis 实现 可以看出, 创建redis 实例实际上是从 pool 中获取 可用的conn 实例绑定到redis 实例上.
redis 实例的 aclose 会将绑定的 conn 实例重置, 然后将 conn 重新放回 pool 的可用 conn 队列中.

但是, 在如果实例可能会被异步或者进程共享, 就可能引发问题.
我们开发习惯于回收连接, 无论是使用 with 或者在 finall 中触发 aclose, 就会导致 redis.connection 变成 None, 从而引发连接被关闭等问题.
所以, 实际上只需要将连接放回pool就可以,不需要额外把 redis.connection 变成 None. 因为重置为 None 本身也没有意义, 因为连接池一直都存在, 内存消耗也不会减少, 反而增加了异常率.

实现方案

从规范上来说, 用过的 redis 实例及时回收, 多个函数之间不要共享实例, 是不会出现这个问题.
但难点在于: 无法确保多个开发之间不会手动回收连接, 所以最好的办法就是重写 aclose.

class SafeRedis(redis.Redis):
    async def aclose(self, *args, **kwargs):
        conn = self.connection
        if conn:
            # 这里注释掉是为了避免在多线程环境下出现连接池泄漏问题
            # self.connection = None
            await self.connection_pool.release(conn)
            logger.info("Redis connection released. not reset self.connection")
class RedisPool:
    def __init__(self):
        self.async_redis_pool = redis.ConnectionPool( ... )

    def client(self) -> redis.Redis:
        return SafeRedis.from_pool(self.async_redis_pool)