Python 单例模式

2019-06-27  本文已影响0人  Oriharas

概述

单例模式(Singleton Pattern)是一种常用的软件设计模式,该模式的主要目的是确保某一个类只有一个实例存在。当你希望在整个系统中,某个类只能出现一个实例时,单例对象就能派上用场。

比如,某个服务器程序的配置信息存放在一个文件中,客户端通过一个 AppConfig 的类来读取配置文件的信息。如果在程序运行期间,有很多地方都需要使用配置文件的内容,也就是说,很多地方都需要创建 AppConfig 对象的实例,这就导致系统中存在多个 AppConfig 的实例对象,而这样会严重浪费内存资源,尤其是在配置文件内容很多的情况下。事实上,类似 AppConfig 这样的类,我们希望在程序运行期间只存在一个实例对象。

使用模块实现单例模式

其实,Python 的模块就是天然的单例模式,因为模块在第一次导入时,会生成 .pyc 文件,当第二次导入时,就会直接加载 .pyc 文件,而不会再次执行模块代码。因此,我们只需把相关的函数和数据定义在一个模块中,就可以获得一个单例对象了。如果我们真的想要一个单例类,可以考虑这样做:

mysingleton.py:

class Singleton(object):
    def foo(self):
        pass
singleton = Singleton()

将上面的代码保存在文件 mysingleton.py 中,要使用时,直接在其他文件中导入此文件中的对象,这个对象即是单例模式的对象:

from a import singleton

使用装饰器

def Singleton(cls):
    _instance = {}

    def _singleton(*args, **kargs):
        if cls not in _instance:
            _instance[cls] = cls(*args, **kargs)
        return _instance[cls]

    return _singleton


@Singleton
class A(object):
    a = 1

    def __init__(self, x=0):
        self.x = x


a1 = A(2)
a2 = A(3)

print("a1: ", a1, "a1.x: ", a1.x)
print("a2: ", a2, "a2.x: ", a2.x)

打印输出:

a1:  <__main__.A object at 0x104433a20> a1.x:  2
a2:  <__main__.A object at 0x104433a20> a2.x:  2

使用类

import threading

class Singleton(object):

    def __init__(self):
        pass

    @classmethod
    def instance(cls, *args, **kwargs):
        if not hasattr(Singleton, "_instance"):
            Singleton._instance = Singleton(*args, **kwargs)
        return Singleton._instance

def task(arg):
    obj = Singleton.instance()
    print(obj)

for i in range(10):
    t = threading.Thread(target=task,args=[i,])
    t.start()

一般情况,大家以为这样就完成了单例模式,但是这样当使用多线程时会存在问题,打印结果如下:

<__main__.Singleton object at 0x10dbbf2b0>
<__main__.Singleton object at 0x10dbbf2b0>
<__main__.Singleton object at 0x10dbbf2b0>
<__main__.Singleton object at 0x10dbbf2b0>
<__main__.Singleton object at 0x10dbbf2b0>
<__main__.Singleton object at 0x10dbbf2b0>
<__main__.Singleton object at 0x10dbbf2b0>
<__main__.Singleton object at 0x10dbbf2b0>
<__main__.Singleton object at 0x10dbbf2b0>
<__main__.Singleton object at 0x10dbbf2b0>

看起来也没有问题,那是因为执行速度过快,如果在 init 方法中有一些 IO 操作,就会发现问题了,下面我们通过time.sleep 模拟,我们在上面 init 方法中加入以下代码:

    def __init__(self):
        import time
        time.sleep(1)

重新执行程序后,结果如下:

<__main__.Singleton object at 0x101f0c2e8>
<__main__.Singleton object at 0x102008630>
<__main__.Singleton object at 0x1020087f0>
<__main__.Singleton object at 0x102008b70>
<__main__.Singleton object at 0x1020088d0>
<__main__.Singleton object at 0x102008a90>
<__main__.Singleton object at 0x102008710>
<__main__.Singleton object at 0x1020089b0>
<__main__.Singleton object at 0x102008550>
<__main__.Singleton object at 0x102008c50>

问题出现了,按照以上方式创建的单例,无法支持多线程。解决办法:加锁。未加锁部分并发执行,加锁部分串行执行,速度降低,但是保证了数据安全:

import threading
import time

class Singleton(object):
    _instance_lock = threading.Lock()

    def __init__(self):
        time.sleep(0.1)

    @classmethod
    def instance(cls, *args, **kwargs):
        with Singleton._instance_lock:
            if not hasattr(Singleton, "_instance"):
                Singleton._instance = Singleton(*args, **kwargs)
        return Singleton._instance

import threading

def task(arg):
    obj = Singleton.instance()
    print(obj)

threads = []
for i in range(10):
    t = threading.Thread(target=task,args=[i,])
    threads.append(t)

[t.start() for t in threads]
[t.join() for t in threads]

obj = Singleton.instance()
print("obj: ", obj)

改造 new 方法

class Singleton:
    instance = None
    init_flag = False

    def __new__(cls, *args, **kwargs):
        if cls.instance is None:
            cls.instance = super().__new__(cls)
        return cls.instance

    def __init__(self, x):
        if Singleton.init_flag:
            return
        self.x = x
        Singleton.init_flag = True

s1 = Singleton(1)
s2 = Singleton(2)
print("s1: ", s1, "s1.x: ", s1.x)
print("s2: ", s2, "s2.x: ", s2.x)

打印输出:

s1:  <__main__.Singleton object at 0x10f9ad940> s1.x:  1
s2:  <__main__.Singleton object at 0x10f9ad940> s2.x:  1

加锁操作代码如下所示:

import threading
import time

class Singleton:
    instance = None
    init_flag = False
    _instance_lock = threading.Lock()
    _instance_lock2 = threading.Lock()
    
    
    def __new__(cls, *args, **kwargs):
        with Singleton._instance_lock:
            if cls.instance is None:
                time.sleep(0.1)
                cls.instance = super().__new__(cls)
            return cls.instance

    def __init__(self, x):
        with Singleton._instance_lock2:
            if Singleton.init_flag:
                return
            self.x = x
            Singleton.init_flag = True

def task(i):
    s = Singleton(i)
    print("s: ", s, "s.x: ", s.x)
    # print("s dir: ", dir(s))

threads = []
for i in range(10):
    t = threading.Thread(target=task, args=(1,)) 
    threads.append(t)

[t.start() for t in threads]
[t.join() for t in threads]
上一篇下一篇

猜你喜欢

热点阅读