单例模式

前言

最近在做一个物联网项目，用户端通过Websocket与服务器进行通信，在服务器上需要将Websocket上的消息通过Mqtt转发给设备，这样就需要在服务器端建立一个Mqtt的通信线程供websocket使用。服务器我采用的是Django+channels来实现Websocket通信。与此同时，我定义了一个MqttClient类用于实现Mqtt收发等功能的实现。该类的通信线程会随Django一同启动。在Webocket收到消息时，我需要通过该进程的对象的send方法将Websocket消息转发至设备。理论来说，一个服务端进程应该保证只启动一个Mqtt的通信线程，否则会导致消息的重复处理。因此我想到了将MqttClient设计成单例模式，保证该类只实例化一个对象，保证通信线程的为唯一。

单例模式

单例模式是一种设计模式。其目的是让类创建的对象在内存中只有唯一的一个实例。在Python中我们可以通过查看类实例化的对象的内存地址，或者通过查看对象的id来区分在不同位置实例化的对象是否为同一个。

实现方法

_new_()方法

Python中，所有的类都会继承Base基类，在Base中定义了__new__()方法，该方法在类的初始化__init()__之前被解释器执行，用于为对象分配内存，返回对象引用。

因此，我们要实现单例模式的思路也就很清晰了，只需要在类分配内存时先检查类是否被实例化，再决定是否分配内存就可以了。

1
2
3
4
5
6
7
8
9
10
11
class Singleton:
    def __new__(cls, *args, **kwargs):
        if not hasattr(cls, "__instance"):
            instance= super(__class__, cls).__new__(cls, *args, **kwargs)
            setattr(cls, "__instance", instance)
        return getattr(cls, "__instance")
    def __init__(self):
        print(self)
if __name__ == "__main__":
    for i in range(0,10):
        Singleton()

运行结果

运行后可以看到实例化的对象地址都相同，说明这些对象仅申请了一次内存。

装饰器方法

在之前的一篇文章我介绍过装饰器的相关原理，下面通过装饰器方式来实现一下单例模式。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def singleton(cls):
    def warpper(*args,**kwargs):
        if hasattr(cls,"__instance"):
            return getattr(cls,"__instance")
        else:
            instance = cls(*args,**kwargs)
            setattr(cls,"__instance",instance)
            return instance
    return warpper
@singleton
class Test:
    def __init__(self):
        print(self)
if __name__ == "__main__":
    for i in range(0,10):
        Test()

运行结果

可以看到，使用装饰器实现单例模式与第一种方法不同，这一次类的构造函数只执行了一次。因为采用装饰器方法时，类还没构造时就已经在外部检查了类的实例化情况，所以类只在第一次进行实例化，以后只返回第一次实例化生成的对象。

线程安全

在多线程下使用单例模式时可能会出现线程安全问题。假如有两个线程，两个线程都实例化一个单例模式类的对象。我们假设一种情况，一个线程先开始实例化，在创建对象时，另一个线程此时正在判断类的实例化情况，或者极端情况下两个类同时进行实例化，那么两个线程在判断实例化情况时都认为该类没有被实例化过，因此各自创建了一个对象，那么这两个线程最后创建了两个不同的对象，单例模式也就失效了。

解决这一问题很简单，只需要加一个线程锁，只有一个线程的实例化结束后另一个线程才会开始判断实例化，这样就避免了线程不安全。