如何在Django中实现Singleton（单例）模式？缓存实例遇序列化错误

我来帮你搞定这个头疼的序列化问题～首先得搞清楚为啥会报错：Redis缓存对象的时候，默认是用pickle来序列化的，但你的单例对象里包含了_thread.lock这种线程锁对象——这类同步原语是和当前进程/线程状态绑定的，pickle根本没法把它们序列化存储，所以才会抛出TypeError: can't pickle _thread.lock objects。

下面给你几个针对性的解决方案，你可以根据自己的场景选：

1. 调整单例实现，把锁移到类级别（推荐） #

很多单例实现会把锁放在实例里，这就导致序列化时把锁也带进去了。你可以把锁移到元类或者类本身的属性里，让实例对象不持有锁：

import threading

class SingletonMeta(type):
    _instances = {}
    # 把锁放在元类里，属于类级别的资源，不会被实例序列化
    _lock = threading.Lock()

    def __call__(cls, *args, **kwargs):
        with cls._lock:
            if cls not in cls._instances:
                cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

# 你的单例类继承这个元类
class YourSingletonClass(metaclass=SingletonMeta):
    def __init__(self):
        # 这里只放业务属性，不要放锁
        self.business_data = "需要缓存的内容"

这样你的单例实例里完全没有锁对象，Redis缓存的时候自然就不会触发序列化错误了。

2. 自定义序列化逻辑，跳过不可序列化的属性 #

如果你的单例必须在实例里持有锁，那可以给类加上__getstate__和__setstate__方法，告诉pickle哪些属性要序列化，哪些要跳过：

import threading

class YourSingleton:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls):
        with cls._lock:
            if not cls._instance:
                cls._instance = super().__new__(cls)
                # 实例化时创建锁
                cls._instance._lock = threading.Lock()
                cls._instance.data = "业务数据"
        return cls._instance

    def __getstate__(self):
        # 复制实例的属性字典，删掉锁这个不可序列化的键
        state = self.__dict__.copy()
        del state['_lock']
        return state

    def __setstate__(self, state):
        # 恢复属性，重新创建锁
        self.__dict__.update(state)
        self._lock = threading.Lock()

这样当Redis用pickle序列化时，会自动调用__getstate__跳过锁；反序列化时__setstate__会重新初始化锁，完全不影响后续使用。

3. 针对你尝试的Django抽象模型单例的优化 #

你写的SingletonModel抽象类，要注意Django模型实例本身可能包含一些内部的不可序列化对象（比如数据库连接相关的状态），直接缓存整个实例很容易出问题。建议不要缓存整个实例，而是缓存模型的字段数据，再按需重构实例：

from django.db import models

class SingletonModel(models.Model):
    class Meta:
        abstract = True

    def save(self, *args, **kwargs):
        # 确保数据库里只有一个实例
        self.__class__.objects.exclude(pk=self.pk).delete()
        super().save(*args, **kwargs)

    @classmethod
    def get_singleton(cls):
        try:
            return cls.objects.get()
        except cls.DoesNotExist:
            instance = cls()
            instance.save()
            return instance

# 缓存的时候只存字段字典，不要存整个实例
singleton = SingletonModel.get_singleton()
cache.set("singleton_key", {k: v for k, v in singleton.__dict__.items() if not k.startswith('_')}, timeout=60*60*24*30)

# 读取缓存时重构实例
cached_data = cache.get("singleton_key")
if cached_data:
    singleton = SingletonModel()
    singleton.__dict__.update(cached_data)
else:
    singleton = SingletonModel.get_singleton()

这里只缓存模型的业务字段（跳过Django内部的下划线开头的属性），取缓存的时候通过字典更新来重构实例，既实现了单例，又避免了序列化问题。

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