深入剖析Python的单例模式实现(二)

发表于:2023-8-31 10:02

字体: | 上一篇 | 下一篇 | 我要投稿

 作者:忆想不到的晖    来源:稀土掘金

  装饰器写法
  重写 __new__ 方法还是比较容易懂,但不太方便使用,每个类都要重写这个方法就很麻烦,逻辑都是一样的,因此我上面抽了一个 BaseSingleton 类来做,通过继承来复用代码。还有一种方式就是通过装饰器来实现单例,把共用的逻辑放到装饰器中做,然后再处理下并发问题。
  def singleton(cls_obj):
      """单例装饰器"""
      _instance_dic = {}
      _instance_lock = threading.Lock()
      @functools.wraps(cls_obj)
      def wrapper(*args, **kwargs):
          if cls_obj in _instance_dic:
              # 实例字典中存在则直接返回
              return _instance_dic.get(cls_obj)
          with _instance_lock:  # 互斥锁,防止多线程竞争,导致创建多实例
              if cls_obj not in _instance_dic:
                  # 实例字典中没有,则创建对象实例,存入字典中
                  _instance_dic[cls_obj] = cls_obj(*args, **kwargs)
          return _instance_dic.get(cls_obj)
      return wrapper
  由于 if cls_obj not in _instance_dic 判断是非原子性操作故而会引发多线程并发问题。
  它大致会转换成以下字节码指令执行:
  1. 加载_instance_dic到栈顶
  Copy code
  LOAD_GLOBAL   0 (_instance_dic)
  2. 加载cls_obj到栈顶
  Copy code
  LOAD_FAST    0 (cls_obj)
  3. 调用__contains__方法检查是否在字典中
  Copy code
  CONTAINS_OP
  4. 根据返回值进行跳转
  Copy code
  POP_JUMP_IF_FALSE  <target>
  如果cls_obj不在_instance_dic中,就会跳转到target位置,也就是if块内的代码。
  可以看到校验是否在字典中是在多个指令中完成,不是一个原子操作。
  在多线程环境下,如果多个线程同时执行到这里,都可能会通过校验,然后创建实例添加到字典中,从而导致线程不安全。
  故而在装饰器中通过线程的互斥锁来解决并发问题,然后通过字典来判断是否存在类的实例对象,存在直接返回,不存在创建对象实例存入字典中来达到单例的效果。
  @singleton
  class Foo(object):
      def __init__(self):
          self.bar = "bar"
  @singleton
  class Demo(object):
      def __init__(self):
          self.demo_name = "singleton_demo"
  f1 = Foo()
  f2 = Foo()
  print(f1)
  print(f2)
  print("f1 is f2", f1 is f2)
  d1 = Demo()
  d2 = Demo()
  print(d1)
  print(d2)
  print("d1 is d2", d1 is d2)
  >>> out
  <__main__.Foo object at 0x102f56c70>
  <__main__.Foo object at 0x102f56c70>
  f1 is f2 True
  <__main__.Demo object at 0x102f56d60>
  <__main__.Demo object at 0x102f56d60>
  d1 is d2 True
  并发安全验证
  @singleton
  class Foo(object):
      def __init__(self):
          self.bar = "bar"
      def two_bar(self):
          return self.bar * 2
  def create_obj(num):
      foo = Foo()
      print(f"foo{num}", foo)
      return foo
  with ThreadPoolExecutor() as pool:
      for i in range(10):
          pool.submit(create_obj, i)
  ok,对象实例都是 <main.Foo object at 0x1016ed700>, 大家可以多运行几次,加了锁不会出现多个实例对象了。
  这里发现被装饰的类都实现了单例模式,接下来我们一探究竟,在装饰器内部打印些东西,看看其工作原理。
  def singleton(cls_obj):
      """单例装饰器"""
      print("cls_obj", cls_obj)
      _instance_dic = {}
      _instance_lock = threading.Lock()
      @functools.wraps(cls_obj)
      def wrapper(*args, **kwargs):
          if cls_obj in _instance_dic:
              # 实例字典中存在则直接返回
              return _instance_dic.get(cls_obj)
          with _instance_lock:  # 互斥锁,防止多线程竞争,导致创建多实例
              if cls_obj not in _instance_dic:
                  # 实例字典中没有,则创建对象实例,存入字典中
                  _instance_dic[cls_obj] = cls_obj(*args, **kwargs)
          print("_instance_dic", _instance_dic, "\n")
          return _instance_dic.get(cls_obj)
      return wrapper
  @singleton
  class Foo(object):
      def __init__(self):
          self.bar = "bar"
  @singleton
  class Demo(object):
      def __init__(self):
          self.demo_name = "singleton_demo"
  f1 = Foo()
  f2 = Foo()
  print(f1)
  print(f2)
  print("f1 is f2", f1 is f2)
  d1 = Demo()
  d2 = Demo()
  print(d1)
  print(d2)
  print("d1 is d2", d1 is d2)
  模块在初始化的时候,其实就会把类初始化形成类对象,注意不是类的实例对象。
  ·装饰器的原理就是python解释器识别到 @singleton 的语法糖时自动把类对象的引用传递给 singleton 装饰器函数
  · 此时装饰器会返回一个新的函数对象(wrapper)出去,把类对象重新赋值了
    -Foo = singleton(Foo) = wrapper
    - Demo = singleton(Demo) = wrapper
  · 到创建对象实例时,Foo() 实则变成了是调用函数 wrapper() 来创建对象
  · 然后每个类都维护了一份 _instance = {} 实例字典,来确保这个类创建的对象只有一份
    - Key 是类对象,eg:Foo、Demo
    - Value 是类的实例对象,eg:Foo(),Demo()
  但装饰器实现的单例模式装饰方便、代码简洁,但是破坏了类的类型,把类变成了函数,导致编写代码的时候没有提示,也不知道有什么属性与方法,所以实际使用起来及其不方便。
  接下来就是引出另一种写法,元类实现单例。
  元类写法
  元类是一种非常晦涩的知识点,一般场景都用不上,但知道元类的原理,后面需要用到时,可以帮助你更好的抽象与封装。
  元类就是创建 类对象的类,type 就是元类
  可以先了解下元类的知识点:追溯Python类的鼻祖——元类 - 掘金
  #!/usr/bin/python3
  # -*- coding: utf-8 -*-
  # @Author: Hui
  # @Desc: { 元类模块 }
  # @Date: 2022/11/26 16:43
  import threading
  class SingletonMetaCls(type):
      """ 单例元类 """
      _instance_lock = threading.Lock()
      def __init__(cls, *args, **kwargs):
          cls._instance = None
          super().__init__(*args, **kwargs)
      def __call__(cls, *args, **kwargs):
          if cls._instance:
              # 存在实例对象直接返回,减少锁竞争,提高性能
              return cls._instance
          with cls._instance_lock:
              if not cls._instance:
                  cls._instance = super().__call__(*args, **kwargs)
          return cls._instance
  使用单例元类进行单例的封装会比装饰器的更好一些,装饰器封装的单例,再实际使用的过程中不太方便,IDE一些开发工具不知道这个类有什么属性,元类就不会,继承也可以实现单例。
  class Foo(metaclass=SingletonMetaCls):
      def __init__(self):
          self.bar = "bar"
      def tow_bar(self):
          return self.bar * 2
  foo1 = Foo()
  foo2 = Foo()
  print("foo1 is foo2", foo1 is foo2)
  >>> out
  foo1 is foo2 True
  继承案例
  class Foo(metaclass=SingletonMetaCls):
      def __init__(self):
          self.bar = "bar"
      def tow_bar(self):
          return self.bar * 2
  foo1 = Foo()
  foo2 = Foo()
  print("foo1 is foo2", foo1 is foo2)
  print("foo1 is foo2", foo1 is foo2)
  class Demo(Foo):
      def __init__(self):
          self.bar = "demo_bar"
  demo1 = Demo()
  demo2 = Demo()
  print("demo1 is demo2", demo1 is demo2)
  print("demo2 two_bar", demo2.tow_bar())
  >>> out
  foo1 is foo2 True
  foo2 two_bar barbar
  demo1 is demo2 True
  demo2 two_bar demo_bardemo_bar
  元类实现原理
  ·加载Foo、Demo等类时,发现指定了元类 metaclass=SingletonMetaCls, 则会让指定的元类来帮助创建类对象
  · 此时 SingletonMetaCls 会调用__init__ 来创建类对象,然后通过super() 让 type 来创建类对象
    -type(类名, 父类元组, 类属性字典)
    - 并动态加了个 cls._instance 属性
  · Foo()、Demo(),创建实例对象时,是Foo、Demo类对象触发了(),所以调用 call() 魔法属性来构造对象实例,存到cls._instance中
  · 下次再创建实例对象,则是先判断是否有,有直接返回,没有则创建
  可以打印一些信息来验证:
  import threading
  class SingletonMetaCls(type):
      """ 单例元类 """
      _instance_lock = threading.Lock()
      def __init__(cls, *args, **kwargs):
          cls._instance = None
          print("SingletonMetaCls __init__", cls)
          print("args", args)
          print("kwargs", kwargs)
          super().__init__(*args, **kwargs)
      def _init_instance(cls, *args, **kwargs):
          if cls._instance:
              # 存在实例对象直接返回,减少锁竞争,提高性能
              print("cls._instance", cls._instance)
              return cls._instance
          with cls._instance_lock:
              if cls._instance is None:
                  cls._instance = super().__call__(*args, **kwargs)
          return cls._instance
      def __call__(cls, *args, **kwargs):
          print("SingletonMetaCls __call__ cur cls", cls)
          instance = cls._init_instance()
          reinit = kwargs.get("reinit", True)
          if reinit:
              # 默认都重新初始化单例对象属性
              instance.__init__(*args, **kwargs)
          return instance
  from py_tools.meta_cls import SingletonMetaCls
  class Foo(metaclass=SingletonMetaCls):
      def __init__(self):
          print("Foo __init__")
          self.bar = "bar"
      def __new__(cls, *args, **kwargs):
          print("Foo __new__")
          return super().__new__(cls, *args, **kwargs)
      def tow_bar(self):
          return self.bar * 2
  # foo1 = Foo()
  # foo2 = Foo()
  # print("foo1 is foo2", foo1 is foo2)
  # print("foo2 two_bar", foo2.tow_bar())
  class Demo(Foo):
      def __init__(self):
          self.bar = "demo_bar"
  模块加载时就会走元类的__init__
  SingletonMetaCls __init__ <class '__main__.Foo'>
  args ('Foo', (), {'__module__': '__main__', '__qualname__': 'Foo', '__init__': <function Foo.__init__ at 0x100eb0310>, '__new__': <function Foo.__new__ at 0x100ed53a0>, 'tow_bar': <function Foo.tow_bar at 0x100ed5430>, '__classcell__': <cell at 0x100eaafd0: SingletonMetaCls object at 0x1217193b0>})
  kwargs {}
  SingletonMetaCls __init__ <class '__main__.Demo'>
  args ('Demo', (<class '__main__.Foo'>,), {'__module__': '__main__', '__qualname__': 'Demo', '__init__': <function Demo.__init__ at 0x100ed54c0>})
  kwargs {}
  看看new对象的时候的打印信息
  class Foo(metaclass=SingletonMetaCls):
      def __init__(self):
          print("Foo __init__")
          self.bar = "bar"
      def __new__(cls, *args, **kwargs):
          print("Foo __new__")
          return super().__new__(cls, *args, **kwargs)
      def tow_bar(self):
          return self.bar * 2
  foo1 = Foo()
  foo2 = Foo()
  print("foo1 is foo2", foo1 is foo2)
  输出信息如下:
  SingletonMetaCls __init__ <class '__main__.Foo'>
  args ('Foo', (), {'__module__': '__main__', '__qualname__': 'Foo', '__init__': <function Foo.__init__ at 0x104998550>, '__new__': <function Foo.__new__ at 0x1049d5550>, 'tow_bar': <function Foo.tow_bar at 0x1049d55e0>, '__classcell__': <cell at 0x104991fd0: SingletonMetaCls object at 0x12f626fb0>})
  kwargs {}
  SingletonMetaCls __call__ cur cls <class '__main__.Foo'>
  Foo __new__
  Foo __init__
  Foo __init__
  SingletonMetaCls __call__ cur cls <class '__main__.Foo'>
  cls._instance <__main__.Foo object at 0x1049b3f70>
  Foo __init__
  foo1 is foo2 True
  四、总结
  ·类重写 new 易懂,但每个类都要重写太冗余了
    -故抽出 BaseSingleton 基类,复用逻辑通过 instance() 来实现单例(推荐)
    - 如果要构造实例属性会有点不太方便
  · 装饰器写法也是复用了创建单例的逻辑,装饰起来方便、简洁
    - 但实际使用装饰过的类不方便,没有类属性提示
  · 元类的写法会有点难与绕,实际使用起来方便,多继承也实现了单例(推荐)
    - 使用起来和平常使用类没有区别
    - 还可以通过reinit参数来控制是否重新初始化实例对象属性
  · 通过线程的互斥锁来解决并发问题
    - 双重判断来减少锁竞争,提高性能
  · 当然还有其他的方式实现单例,例如通过Python的模块导入,来保证只会创建一个实例。
  本文内容不用于商业目的,如涉及知识产权问题,请权利人联系51Testing小编(021-64471599-8017),我们将立即处理
《2023软件测试行业现状调查报告》独家发布~

关注51Testing

联系我们

快捷面板 站点地图 联系我们 广告服务 关于我们 站长统计 发展历程

法律顾问:上海兰迪律师事务所 项棋律师
版权所有 上海博为峰软件技术股份有限公司 Copyright©51testing.com 2003-2024
投诉及意见反馈:webmaster@51testing.com; 业务联系:service@51testing.com 021-64471599-8017

沪ICP备05003035号

沪公网安备 31010102002173号