封装redis-py时出现连接拒绝异常的问题求助

问题场景 #

直接创建Redis实例可正常执行ping()：

>>> r = redis.Redis(host='192.168.1.xxx', port=xxxx, decode_responses=True)
>>> r.ping()
True

但封装为工具类后调用ping()抛出连接拒绝错误：

import redis

class RedisCommunication:
    def __init__(self, host=None, port=None):
        self.REDIS_HOST = '192.168.1.xxx'
        self.REDIS_PORT = xxxx

        try:
            self.rc = redis.Redis(
                host=self.REDIS_HOST,
                port=self.REDIS_PORT,
                db=0,
                decode_responses=True)
        except redis.ConnectionError:
            exit(1)

# 调用代码
>>> redcom = RedisCommunication()
>>> redcom.rc.ping()

错误栈信息：

Traceback (most recent call last):
  File "/opt/homebrew/Caskroom/mambaforge/base/envs/wadjet-backend-env/lib/python3.11/site-packages/redis/connection.py", line 699, in connect
    sock = self.retry.call_with_retry(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Caskroom/mambaforge/base/envs/wadjet-backend-env/lib/python3.11/site-packages/redis/retry.py", line 46, in call_with_retry
    return do()
           ^^^^
  File "/opt/homebrew/Caskroom/mambaforge/base/envs/wadjet-backend-env/lib/python3.11/site-packages/redis/connection.py", line 700, in <lambda>
    lambda: self._connect(), lambda error: self.disconnect(error)
            ^^^^^^^^^^^^^^^
  File "/opt/homebrew/Caskroom/mambaforge/base/envs/wadjet-backend-env/lib/python3.11/site-packages/redis/connection.py", line 1002, in _connect
    raise err
  File "/opt/homebrew/Caskroom/mambaforge/base/envs/wadjet-backend-env/lib/python3.11/site-packages/redis/connection.py", line 990, in _connect
    sock.connect(socket_address)
ConnectionRefusedError: [Errno 61] Connection refused
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/Users/xxx/Library/Application Support/JetBrains/Toolbox/apps/PyCharm-P/ch-0/231.9011.38/PyCharm.app/Contents/plugins/python/helpers/pydev/pydevconsole.py", line 364, in runcode
    coro = func()
           ^^^^^^
  File "<input>", line 1, in <module>
  File "/opt/homebrew/Caskroom/mambaforge/base/envs/wadjet-backend-env/lib/python3.11/site-packages/redis/commands/core.py", line 1205, in ping
    return self.execute_command("PING", **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Caskroom/mambaforge/base/envs/wadjet-backend-env/lib/python3.11/site-packages/redis/client.py", line 1266, in execute_command
    conn = self.connection or pool.get_connection(command_name, **options)
                              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/Caskroom/mambaforge/base/envs/wadjet-backend-env/lib/python3.11/site-packages/redis/connection.py", line 1457, in get_connection
    connection.connect()
  File "/opt/homebrew/Caskroom/mambaforge/base/envs/wadjet-backend-env/lib/python3.11/site-packages/redis/connection.py", line 705, in connect
    raise ConnectionError(self._error_message(e))
redis.exceptions.ConnectionError: Error 61 connecting to 192.168.1.xxx:xxxx. Connection refused.

核心原因分析 #

1. redis-py延迟连接机制：redis-py默认不会在实例化Redis对象时立即建立连接，而是延迟到执行第一个命令（比如ping()）时才发起连接请求。因此工具类__init__中的try-except块根本不会触发——此时还没尝试连接，自然不会抛出ConnectionError。

2. 配置不一致可能性：工具类中硬编码的REDIS_HOST/REDIS_PORT可能和直接实例化时使用的参数不一致，比如IP拼写错误、端口号写错，导致真正发起连接时失败。

3. 参数逻辑错误：工具类__init__定义了host/port形参，但完全没使用，而是硬编码固定值，若调用时传入了不同的配置，也会导致连接错误。

修复方案 #

方案1：验证并统一配置 #

确保工具类中硬编码的REDIS_HOST和REDIS_PORT与直接实例化时使用的参数完全一致，检查是否有拼写、数字错误。

方案2：支持外部参数传入 #

修改工具类逻辑，优先使用外部传入的配置，没有则用默认值：

import redis

class RedisCommunication:
    def __init__(self, host='192.168.1.xxx', port=xxxx):
        self.REDIS_HOST = host
        self.REDIS_PORT = port

        self.rc = redis.Redis(
            host=self.REDIS_HOST,
            port=self.REDIS_PORT,
            db=0,
            decode_responses=True)

    def ping(self):
        try:
            return self.rc.ping()
        except redis.ConnectionError:
            print("Redis连接失败")
            return False

方案3：初始化时主动验证连接 #

如果需要在类初始化阶段就确认连接状态，主动调用ping()触发连接：

import redis

class RedisCommunication:
    def __init__(self, host='192.168.1.xxx', port=xxxx):
        self.REDIS_HOST = host
        self.REDIS_PORT = port

        self.rc = redis.Redis(
            host=self.REDIS_HOST,
            port=self.REDIS_PORT,
            db=0,
            decode_responses=True)

        # 初始化时验证连接
        try:
            self.rc.ping()
        except redis.ConnectionError:
            print("Redis初始化连接失败")
            exit(1)

内容的提问来源于stack exchange，提问作者Sam