Kubernetes中Redis Pod重启时redis-py ConnectionError的处理

针对你遇到的Kubernetes上Redis Pod频繁重启/迁移，导致redis-py抛出ConnectionError、python-rq任务中断的问题，结合场景给出以下实际优化方向：

一、优化redis-py连接池配置 #

• 缩短健康检查间隔：当前health_check_interval=300（5分钟）刚好覆盖Pod恢复时间，但调小到30秒能让连接池更快发现失效连接——每次借出连接时，会自动对超过间隔的连接发送PING命令，直接重建死连接，避免拿到无效连接报错。
• 补充重试错误类型：Redis重启加载持久化文件时会抛出BusyLoadingError，这个也需要加入retry_on_error列表，避免加载期间的连接失败直接中断流程。
• 开启TCP保活：加上socket_keepalive=True，配合系统TCP保活参数，能更早检测到连接断开，避免客户端长期持有无效连接。
• 可选：调整重试策略：如果想初期快速重试、后期延长间隔，可把ConstantBackoff换成ExponentialBackoff，比如ExponentialBackoff(initial=1, exponent=2, max=10)，既覆盖Pod恢复窗口，又减少不必要的频繁重试。

调整后的代码示例：

from redis.retry import Retry
from redis.backoff import ConstantBackoff, ExponentialBackoff
from redis.exceptions import TimeoutError, ConnectionError, NoPermissionError, BusyLoadingError
import redis


def get_redis_connection():
    redis_pool = redis.ConnectionPool(
        host="XXXX",
        port=1234,
        # 可选：用指数退避替代固定间隔
        # retry=Retry(ExponentialBackoff(initial=1, max=10), 30),
        retry=Retry(ConstantBackoff(10), 30),
        retry_on_error=[
            ConnectionError, TimeoutError, NoPermissionError, 
            ConnectionRefusedError, PermissionError, BusyLoadingError
        ],
        socket_timeout=300,
        socket_connect_timeout=300,
        health_check_interval=30,  # 缩短健康检查间隔
        socket_keepalive=True,  # 开启TCP保活
    )
    return redis.Redis(connection_pool=redis_pool)

二、适配python-rq的任务重试逻辑 #

redis-py的客户端重试只覆盖单次命令执行，但python-rq任务可能在执行中途遇到连接中断，需要让rq本身具备重试能力：

• 任务级重试配置：定义任务时，用@job装饰器指定连接异常的重试规则：

  from rq import job
  from redis.exceptions import ConnectionError
  
  @job('default', retry=3, retry_on=[ConnectionError])
  def my_task():
      # 任务业务逻辑
  

• 全局Worker异常处理器：给rq Worker添加全局处理器，捕获连接错误后触发任务重试：

  from rq import Worker, Queue, Connection
  from redis.exceptions import ConnectionError
  
  def handle_connection_error(job, exc_type, exc_value, traceback):
      if exc_type == ConnectionError:
          job.meta.setdefault('retry_count', 0)
          job.meta['retry_count'] += 1
          if job.meta['retry_count'] < 3:
              job.save_meta()
              return True  # 返回True表示继续重试任务
      return False
  
  with Connection(get_redis_connection()):
      worker = Worker([Queue('default')], exception_handlers=[handle_connection_error])
      worker.work()
  

• 强制使用自定义连接池：必须将你配置好的带重试/健康检查的连接实例传入rq的Connection，不要用rq默认的简单Redis连接，确保Worker所有操作都复用优化后的连接池。

三、Kubernetes层面辅助优化 #

• 确认Service配置：确保Redis的Service为ClusterIP类型，Kubernetes会自动更新Service的Endpoint指向新Pod。如果客户端存在DNS缓存问题，redis-py的健康检查和连接超时配置会强制每次新建连接时重新解析DNS。
• 完善Redis Pod探针：给Redis Pod配置livenessProbe和readinessProbe，比如用redis-cli ping作为探测命令，确保Pod真正就绪后才加入Service Endpoint，避免客户端连接到启动中的Pod。

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