黄兵的个人博客

Python Redis 单例模式

这是一个通过单例模式连接 Redis 数据库的代码：

class RedisOperation(object):
    def __new__(cls):
        # 使用单例模式连接 Redis
        if not hasattr(cls, 'instance'):
            pool = redis.ConnectionPool(host=current_app.config['REDIS_HOST'],
                                        port=int(current_app.config['REDIS_PORT']),
                                        password=current_app.config['REDIS_PASSWORD'],
                                        decode_responses=current_app.config['DECODE_RESPONSES'])
            cls.instance = redis.Redis(connection_pool=pool)
        return cls.instance

这段代码使用了 单例模式 来管理 Redis 连接池，主要目的是确保所有 Redis 操作都使用同一个实例，以减少连接开销。但是，这样的实现方式有以下 几个问题：

问题 1：__new__ 直接返回 redis.Redis 实例

现象

• __new__ 方法返回的是 redis.Redis 实例，而不是 RedisOperation 实例。

• 这意味着 RedisOperation 这个类实际上无法调用自己的方法，因为 RedisOperation() 返回的是 redis.Redis 实例，而 redis.Redis 不会有 RedisOperation 可能定义的其他方法。

示例

redis_op = RedisOperation()
print(type(redis_op))  # class 'redis.client.Redis'

# 假设 RedisOperation 里面定义了一个自定义方法
class RedisOperation:
    def custom_method(self):
        return "Hello"

redis_op = RedisOperation()
print(redis_op.custom_method())  # AttributeError: 'Redis' object has no attribute 'custom_method'

问题 2：使用 current_app.config 可能导致 ImportError

现象

current_app 只能在 Flask 应用上下文 内使用，而 RedisOperation 可能会在应用上下文之外被导入。例如，在 Flask 之外的地方（如单元测试、任务队列等）导入 RedisOperation，会导致 RuntimeError: Working outside of application context 错误。

解决方案

可以使用 flask.current_app 进行懒加载，或者在 __init__ 时传入 Redis 配置，而不是直接依赖 current_app.config。

问题 3：类属性 instance 不是线程安全的

现象

• hasattr(cls, 'instance') 不是原子操作，可能导致多个线程同时判断 instance 不存在，然后都创建了新实例，导致 多个 Redis 连接 被创建。

• 这可能会带来 多个 Redis 连接池，从而影响 Redis 连接管理。

解决方案

• 使用 threading.Lock 进行线程同步

• 使用 redis.ConnectionPool 让所有 redis.Redis 实例共享同一个连接池

优化后的实现

import redis
import threading
from flask import current_app

class RedisOperation:
    _instance = None
    _lock = threading.Lock()  # 确保线程安全

    def __new__(cls):
        if not cls._instance:
            with cls._lock:  # 线程锁，防止多个线程同时创建实例
                if not cls._instance:  # 双重检查锁定（Double-Checked Locking）
                    pool = redis.ConnectionPool(
                        host=current_app.config.get('REDIS_HOST', 'localhost'),
                        port=int(current_app.config.get('REDIS_PORT', 6379)),
                        password=current_app.config.get('REDIS_PASSWORD', None),
                        decode_responses=current_app.config.get('DECODE_RESPONSES', True)
                    )
                    cls._instance = redis.Redis(connection_pool=pool)
        return cls._instance

优化点

✅ 修正 __new__ 逻辑

• 现在 RedisOperation() 返回的仍然是 RedisOperation 类的实例，而不是 redis.Redis 。

• 但 cls._instance 是 redis.Redis，确保 Redis 连接共享。

✅ 避免 current_app 直接初始化

• 通过 current_app.config.get() 方式，在没有 Flask 上下文时不会抛出异常，而是使用默认值。

✅ 线程安全

• 使用 threading.Lock 防止多线程创建多个 Redis 连接。

✅ 使用 redis.ConnectionPool

• 连接池可以确保多个 RedisOperation 实例共享同一个 Redis 连接，提高性能并减少资源消耗。

示例使用

redis_client = RedisOperation()
redis_client.set("foo", "bar")

print(redis_client.get("foo"))  # "bar"

这个实现更加安全、可扩展，推荐使用！🚀