7

Python 设计模式——单例模式

 3 years ago
source link: https://rollingstarky.github.io/2020/11/27/python-design-patterns-singleton/
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
neoserver,ios ssh client

Python 设计模式——单例模式

2020-11-27

| Python

|

|

4.6k

|

0:05

单例模式即确保类有且只有一个特定类型的对象,并提供全局访问点。因此通常用于日志记录、数据库操作、打印机后台处理程序等。这些程序在运行过程中只生成一个实例,避免对同一资源产生相互冲突的请求。

特点:

  • 确保类有且只有一个对象被创建
  • 为唯一对象提供访问点,令其可被全局访问
  • 控制共享资源的并行访问

经典单例模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class Singleton(object):
def __new__(cls, name):
if not hasattr(cls, 'instance'):
cls.instance = super().__new__(cls)
return cls.instance

def __init__(self, name):
self.name = name


s1 = Singleton('Singleton1')
print(s1)
# => <__main__.Singleton object at 0x7efc1b006220>
print(s1.name)
# => Singleton1

s2 = Singleton('Singleton2')
print(s2)
# => <__main__.Singleton object at 0x7efc1b006220>
print(s2.name)
# => Singleton2
print(s1.name)
# => Singleton2

在上面的代码中,通过定义 __new__ 方法控制对象的创建。方法 hasattr 则用于检查对象 cls 是否具有 instance 属性(即确认该类是否已经生成了一个对象)。若 instance 属性不存在,则使用 super().__new__() 方法创建新的实例;若 instance 属性存在,则分配已有的实例给变量。
因此当 s2 = Singleton('Singleton2') 执行时,hasattr 发现对象实例已存在(s1),因此直接将已有的对象分配给 s2。s1 和 s2 实际是同一个对象实例。

Monostate(单态)模式

Monostate 模式即类的所有实例对象共享相同的状态。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class Borg:
__shared_state = {}
def __init__(self):
self.__dict__ = self.__shared_state

b = Borg()
b1 = Borg()

print(b is b1) # => False
b.x = 4
print(b.x) # => 4
print(b1.x) # => 4
b1.x = 6
print(b1.x) # => 6
print(b.x) # => 6

在上述代码中,通过将类变量 __shared_state 赋值给实例变量 __dict____dict__ 变量用于存储实例对象的属性等状态),使得类生成的所有对象实例都共享同一状态。
即 b 和 b1 是 Borg 类创建的不同的实例对象,但用于保存实例状态的 b.__dict__b1.__dict__ 却是相同的(即都是 Borg.__shared_state)。因此 b 的属性 x 若发生改变,同样的变化也会体现到 b1 中。

也可以通过修改 __new__ 方法来实现 Borg 模式:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class Borg:
__shared_state = {}
def __new__(cls, name):
obj = super().__new__(cls)
obj.__dict__ = cls.__shared_state
return obj

def __init__(self, name):
self.name = name


b1 = Borg('Borg1')
print(b1.name) # => Borg1
b2 = Borg('Borg2')
print(b2.name) # => Borg2
print(b1.name) # => Borg2
b1.name = 'Borg'
print(b1.name) # => Borg
print(b2.name) # => Borg
print(b1 is b2) # => False

通过元类实现单例模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class MetaSingleton(type):
def __init__(self, *args, **kwargs):
self.__instance = None

def __call__(self, *args, **kwargs):
if not self.__instance:
self.__instance = super().__call__(*args, **kwargs)
return self.__instance


class Logger(metaclass=MetaSingleton):
pass


logger1 = Logger()
logger2 = Logger()
print(logger1, logger2)
# => <__main__.Logger object at 0x7fac8af577c0> <__main__.Logger object at
# 0x7fac8af577c0>
print(logger1 is logger2) # => True

单例模式的实际应用

DB 操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import sqlite3

class MetaSingleton(type):
def __init__(self, *args, **kwargs):
self.__instance = None

def __call__(self, *args, **kwargs):
if not self.__instance:
self.__instance = super().__call__(*args, **kwargs)
return self.__instance


class Database(metaclass=MetaSingleton):
connection = None
def connect(self):
if self.connection is None:
self.connection = sqlite3.connect("db.sqlite3")
self.cursorobj = self.connection.cursor()
return self.cursorobj


db1 = Database().connect()
db2 = Database().connect()

print(db1, db2)
# => <sqlite3.Cursor object at 0x7f810d6f8260> <sqlite3.Cursor object at
# 0x7f810d6f8260>
print(db1 is db2)
# => True

监控服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class HealthCheck:
_instance = None
def __new__(cls, *args, **kwargs):
if not HealthCheck._instance:
HealthCheck._instance = super().__new__(cls, *args, **kwargs)
return HealthCheck._instance

def __init__(self):
self._servers = []

def addServer(self):
self._servers.append("Server 1")
self._servers.append("Server 2")
self._servers.append("Server 3")
self._servers.append("Server 4")

def changeServer(self):
self._servers.pop()
self._servers.append("Server 5")

hc1 = HealthCheck()
hc2 = HealthCheck()

hc1.addServer()
print("Schedule health check for servers (1) ...")
for i in range(4):
print("Checking ", hc1._servers[i])

hc2.changeServer()
print("Schedule health check for servers (2) ...")
for i in range(4):
print("Checking ", hc2._servers[i])

# => Schedule health check for servers (1) ...
# => Checking Server 1
# => Checking Server 2
# => Checking Server 3
# => Checking Server 4
# => Schedule health check for servers (2) ...
# => Checking Server 1
# => Checking Server 2
# => Checking Server 3
# => Checking Server 5



About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK