【零】引入#

【1】前言#

我们知道，经典设计模式总共有 23 种，但其中只有少数几种被广泛采用。
根据我的工作经验，实际常用的可能不超过其中的一半。
如果随机找一位程序员，并要求他列举出自己最熟悉的三种设计模式，那么单例模式肯定会是其中之一，这也是今天我们要讨论的。

【2】为什么要单例模式#

单例设计模式（Singleton Design Pattern）: 一个类只允许创建一个对象（或者实例），那这个类就是一个单例类，这种设计模式就叫作单例设计模式，简称单例模式。
当一个类的功能比较单一，只需要一个实例对象就可以完成需求时，就可以使用单例模式来节省内存资源。

【通常】单例模式创建的对象是进程唯一的, 单例类中对象的唯一性的作用范围是进程内的，在进程间是不唯一的。

【3】如何实现一个单例#

要实现一个单例，我们需要知道要重点关注的点是哪些？
- 考虑对象创建时的线程安全问题
- 考虑是否支持延迟加载
- 考虑获取实例的性能是否高（是否加锁）
在python中，我们可以使用多种方法来实现单例模式
- 使用模块
- 使用装饰器
- 使用类（方法）
- 基于__new__方法实现
- 基于元类metaclass实现

【4】饿汉式与懒汉式#

（1）饿汉式(着急吃)#

饿汉式的实现方式比较简单。在类加载的时候，instance 静态实例就已经创建并初始化好了，所以，instance 实例的创建过程是线程安全的。
不过，这样的实现方式不支持延迟加载（在真正用到 IdGenerator 的时候，再创建实例）。
一种常见的方式是利用python模块化实现单例。

1
import threading
2

3

4
class IdGenerator:
5
    def __init__(self):
6
        self._lock = threading.Lock()
7
        self.id = 0
8

9
    def get_id(self):
10
        with self._lock:
11
            self.id += 1
12
            return self.id
13

14

15
# 创建单例实例
16
id_generator_instance = IdGenerator()

在另一个文件（例如 main.py）中导入模块并使用单例模式

1
import id_generator_module
2

3
if __name__ == '__main__':
4
    singleton1 = id_generator_module.id_generator_instance
5
    singleton2 = id_generator_module.id_generator_instance
6

7
    print(singleton1 is singleton2)  # 输出 True，表示是同一个实例

Python中可以使用模块的方式非常简单地实现单例模式，因为Python模块在程序中只会被导入一次，所以它天生就是单例的。

（2）懒汉式（不着急吃）#

有饿汉式，对应的，就有懒汉式。懒汉式相对于饿汉式的优势是支持延迟加载。
- 使用装饰器
- 使用类（方法）
- 基于__new__方法实现
- 基于元类metaclass实现

【一】类属性#

【1】类产生对象#

当我们有一个类，类里面有很多方法
当我们想使用这些方法的时候，我们的第一想法是实例化得到一个对象，再用对象去调用相应的方法
但是不同的人，都想用这一个方法的时候，就需要每一个人都去实例化对象，用对象调用方法
但是我们不去使用对象中的其他属性，只用到其中一种属性，于是这对于系统来说是一种开销

1
# 创建一个普通的类
2
class MysqlControl(object):
3
    pass
4

5

6
# 实例化类得到对象 --- 类只要加 () 实例化就会产生一个全新的对象
7
obj_one = MysqlControl()
8
obj_two = MysqlControl()
9

10
# 查看对象  --- 发现虽然是同一个类实例化得到的对象，但是对象的地址不一样
11
print(obj_one)
12
# <__main__.MysqlControl object at 0x00000204C2BB3FD0>
13
print(obj_two)
14
# <__main__.MysqlControl object at 0x00000204C2BB3F40>
15
print(obj_one is obj_two)
16
# False

【2】类属性包装成方法#

1
class Singleton(object):
2
    _instance = None
3

4
    def __init__(self, ip, port):
5
        self.ip = ip
6
        self.port = port
7

8
    @classmethod
9
    def get_instance(cls):
10
        if cls._instance is None:
11
            cls._instance = Singleton('127.0.0.1', 3306)
12
        return cls._instance
13

14

15
obj_one = Singleton.get_instance()
16
obj_two = Singleton.get_instance()
17

18
print(obj_one)
19
# <__main__.Singleton object at 0x0000023CA8B03E50>
20
print(obj_two)
21
# <__main__.Singleton object at 0x0000023CA8B03E50>
22
print(obj_one is obj_two)
23
# True
24
# 输出 True，表示是同一个实例

使用类属性保存实例，通过类方法获取实例。
在第一次调用get_instance方法时创建实例，并在后续调用中直接返回该实例。

【二】装饰器#

1
def singleton(cls):
2
    instances = {}
3

4
    def wrapper(*args, **kwargs):
5
        if cls not in instances:
6
            instances[cls] = cls(*args, **kwargs)
7
        return instances[cls]
8

9
    return wrapper
10

11

12
@singleton
13
class Singleton:
14
    pass
15

16

17
singleton_one = Singleton()
18
singleton_two = Singleton()
19

20
print(singleton_one)
21
# <__main__.Singleton object at 0x000001C0B531A770>
22
print(singleton_two)
23
# <__main__.Singleton object at 0x000001C0B531A770>
24
print(singleton_one is singleton_two)
25
# True
26

27
# 输出 True，表示是同一个实例
28

29
# # 使用 __call__ 方法获取单例
30
singleton_three = Singleton()
31
print(singleton_one is singleton_three)
32
# True

使用装饰器将原来的类包装成一个新的类，通过闭包和字典保存实例。
在每次实例化时，先检查字典中是否已经存在该类的实例，如果不存在才创建实例并返回。

【三】元类(metaclass)#

1
class SingletonType(type):
2
    def __init__(cls, name, bases, attrs):
3
        super(SingletonType, cls).__init__(name, bases, attrs)
4
        cls.instance = None
5

6
    def __call__(cls, *args, **kwargs):
7
        if cls.instance is None:
8
            cls.instance = super(SingletonType, cls).__call__(*args, **kwargs)
9
        return cls.instance
10

11

12
class Singleton(metaclass=SingletonType):
13
    pass
14

15

16
singleton_one = Singleton()
17
singleton_two = Singleton()
18

19
print(singleton_one)
20
# <__main__.Singleton object at 0x000001C0B531A770>
21
print(singleton_two)
22
# <__main__.Singleton object at 0x000001C0B531A770>
23
print(singleton_one is singleton_two)
24
# True
25

26
# 输出 True，表示是同一个实例
27

28
# # 使用 __call__ 方法获取单例
29
singleton_three = Singleton()
30
print(singleton_one is singleton_three)
31
# True

定义一个元类，在元类的__call__方法中判断实例是否已存在，如果不存在则调用父类的__call__方法来创建并返回实例。

【四】基于new方法#

【1】思路#

在Python中，对象的实例化过程通常遵循以下步骤：
- 首先，执行类的__new__方法，如果未定义此方法，将默认调用父类的__new__方法来创建一个实例化对象。
- 接着，再执行__init__方法来对这个新创建的对象进行初始化。
我们可以充分利用这个实例化过程来实现单例模式。
- 具体做法是在类的__new__方法中判断是否已经存在实例，如果存在，则直接返回现有的实例，否则创建一个新的实例。
- 这样就能够确保只有一个实例存在，从而实现了单例模式的效果。

【2】代码#

1
class Singleton(object):
2
    def __new__(cls, *args, **kwargs):
3
        if not hasattr(cls, '_instance'):
4
            cls._instance = super().__new__(cls)
5
        return cls._instance
6

7

8
singleton_one = Singleton()
9
singleton_two = Singleton()
10

11
print(singleton_one)
12
# <__main__.Singleton object at 0x000001C0B531A770>
13
print(singleton_two)
14
# <__main__.Singleton object at 0x000001C0B531A770>
15
print(singleton_one is singleton_two)
16
# True
17

18
# 输出 True，表示是同一个实例
19

20
# # 使用 __call__ 方法获取单例
21
singleton_three = Singleton()
22
print(singleton_one is singleton_three)
23
# True

重写__new__方法，在实例化对象时判断类中是否已有实例，如果没有则调用父类的__new__方法来创建并返回。

【3】优势#

这种实现方式会导致频繁加锁、释放锁，以及并发度低等问题，频繁的调用会产生性能瓶颈。

【五】基于模块#

1
class Singleton:
2
    pass
3

4
singleton_instance = Singleton()

将实例化操作放在模块级别，通过导入该模块来获取实例。
由于Python模块在运行时只会被导入一次，因此保证了实例的单一性。

【使用场景】#

【1】配置信息#

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

1
import configparser
2
import threading
3

4

5
class Config:
6
    _instance = None
7
    _lock = threading.Lock()
8

9
    def __new__(cls):
10
        if cls._instance is None:
11
            with cls._lock:
12
                if cls._instance is None:
13
                    cls._instance = super(Config, cls).__new__(cls)
14
                    cls._instance.load_config()
15
        return cls._instance
16

17
    def load_config(self):
18
        self.config = configparser.ConfigParser()
19
        if not self.config.read('config.ini'):
20
            # 如果配置文件不存在，设置默认配置
21
            self.set_default_config()
22

23
    def set_default_config(self):
24
        # 设置默认配置
25
        self.config['Section1'] = {'key1': 'default_value1', 'key2': 'default_value2'}
26

27
    def get_config(self, section, key):
28
        return self.config.get(section, key)
29

30
    def update_config(self, section, key, value):
31
        self.config.set(section, key, value)
32
        with open('config.ini', 'w') as config_file:
33
            self.config.write(config_file)
34

35

36
if __name__ == "__main__":
37
    config = Config()
38

39
    # 获取配置信息
40
    value1 = config.get_config("Section1", "key1")
41
    print(value1)
42

43
    # 更新配置信息
44
    config.update_config("Section1", "key1", "new_value")
45

46
    # 获取更新后的配置信息
47
    updated_value1 = config.get_config("Section1", "key1")
48
    print(updated_value1)

【2】分布式ID(雪花算法)#

分布式ID生成是一个常见的需求，以下是一个使用雪花算法实现分布式ID生成的Python代码示例，并将雪花算法的生成ID功能与单例模式结合使用，创建了一个单例类，该类包含了雪花算法的实例，并确保只有一个该类的实例存在

1
import threading
2
import time
3

4

5
class SnowflakeIDGenerator:
6
    def __init__(self, worker_id, datacenter_id):
7
        # 41位时间戳位
8
        self.timestamp_bits = 41
9
        # 10位工作机器ID位
10
        self.worker_id_bits = 10
11
        # 12位序列号位
12
        self.sequence_bits = 12
13

14
        # 最大工作机器ID和最大序列号
15
        self.max_worker_id = -1 ^ (-1 << self.worker_id_bits)
16
        self.max_sequence = -1 ^ (-1 << self.sequence_bits)
17

18
        # 时间戳左移的位数
19
        self.timestamp_shift = self.worker_id_bits + self.sequence_bits
20
        # 工作机器ID左移的位数
21
        self.worker_id_shift = self.sequence_bits
22

23
        # 配置工作机器ID和数据中心ID
24
        self.worker_id = worker_id
25
        self.datacenter_id = datacenter_id
26

27
        # 初始化序列号
28
        self.sequence = 0
29
        # 上次生成ID的时间戳
30
        self.last_timestamp = -1
31

32
        # 线程锁，用于保护并发生成ID的安全性
33
        self.lock = threading.Lock()
34

35
        # 校验工作机器ID和数据中心ID是否合法
36
        if self.worker_id < 0 or self.worker_id > self.max_worker_id:
37
            raise ValueError(f"Worker ID must be between 0 and {self.max_worker_id}")
38
        if self.datacenter_id < 0 or self.datacenter_id > self.max_worker_id:
39
            raise ValueError(f"Datacenter ID must be between 0 and {self.max_worker_id}")
40

41
    def _current_timestamp(self):
42
        return int(time.time() * 1000)
43

44
    def _wait_for_next_timestamp(self, last_timestamp):
45
        timestamp = self._current_timestamp()
46
        while timestamp <= last_timestamp:
47
            timestamp = self._current_timestamp()
48
        return timestamp
49

50
    def generate_id(self):
51
        with self.lock:
52
            current_timestamp = self._current_timestamp()
53
            if current_timestamp < self.last_timestamp:
54
                raise ValueError("Clock moved backwards. Refusing to generate ID.")
55

56
            if current_timestamp == self.last_timestamp:
57
                self.sequence = (self.sequence + 1) & self.max_sequence
58
                if self.sequence == 0:
59
                    current_timestamp = self._wait_for_next_timestamp(self.last_timestamp)
60
            else:
61
                self.sequence = 0
62

63
            self.last_timestamp = current_timestamp
64

65
            # 构造ID
66
            timestamp = current_timestamp << self.timestamp_shift
67
            worker_id = self.worker_id << self.worker_id_shift
68
            id = timestamp | worker_id | self.sequence
69
            return id
70

71

72
class SingletonSnowflakeGenerator:
73
    _instance_lock = threading.Lock()
74
    _instance = None
75

76
    def __new__(cls, worker_id, datacenter_id):
77
        if cls._instance is None:
78
            with cls._instance_lock:
79
                if cls._instance is None:
80
                    cls._instance = SnowflakeIDGenerator(worker_id, datacenter_id)
81
        return cls._instance
82

83

84
if __name__ == "__main__":
85
    generator1 = SingletonSnowflakeGenerator(worker_id=1, datacenter_id=1)
86
    generator2 = SingletonSnowflakeGenerator(worker_id=2, datacenter_id=2)
87

88
    print(generator1 is generator2)  # 输出 True，表示是同一个实例
89

90
    id1 = generator1.generate_id()
91
    id2 = generator2.generate_id()
92

93
    print(id1)  # 7108005303425175552
94
    print(id2)  # 7108005303425175553

【3】数据库连接池#

确保在应用程序中只存在一个数据库连接池的实例，以提高性能和资源利用率。

1
import threading
2
import pymysql
3
from dbutils.pooled_db import PooledDB
4

5

6
class DatabaseConnectionPoolProxy:
7
    _instance_lock = threading.Lock()
8

9
    def __new__(cls, *args, **kwargs):
10
        if not hasattr(DatabaseConnectionPoolProxy, "_instance"):
11
            with DatabaseConnectionPoolProxy._instance_lock:
12
                if not hasattr(DatabaseConnectionPoolProxy, "_instance"):
13
                    DatabaseConnectionPoolProxy._instance = object.__new__(cls)
14
                    cls._instance.initialize_pool()
15
        return DatabaseConnectionPoolProxy._instance
16

17
    def initialize_pool(self):
18
        self.pool = PooledDB(
19
            creator=pymysql,
20
            maxconnections=6,
21
            mincached=2,
22
            maxcached=5,
23
            maxshared=3,
24
            blocking=True,
25
            maxusage=None,
26
            # 配置其他数据库连接池参数
27
            host='192.168.91.1',
28
            port=3306,
29
            user='root',
30
            password='root',
31
            database='inventory',
32
            charset='utf8'
33
        )
34

35
    def get_connection(self):
36
        if self.pool:
37
            return self.pool.connection()
38

39
    def execute_query(self, query, params=None):
40
        conn = self.get_connection()
41
        if conn:
42
            cursor = conn.cursor()
43
            try:
44
                cursor.execute(query, params)
45
                result = cursor.fetchall()
46
                return result
47
            finally:
48
                cursor.close()
49
                conn.close()
50

51

52
if __name__ == "__main__":
53
    db_proxy = DatabaseConnectionPoolProxy()
54
    result = db_proxy.execute_query("SELECT * FROM inventory WHERE id=%s", [5])
55

56
    print(result)
57

58
    db_proxy1 = DatabaseConnectionPoolProxy()
59
    db_proxy2 = DatabaseConnectionPoolProxy()
60

61
    print(db_proxy1 is db_proxy2)  # True

【4】缓存管理#

管理应用程序中的缓存数据，确保只有一个缓存管理器实例来避免数据一致性问题。

1
import threading
2

3

4
class CacheManager:
5
    _instance = None
6
    _lock = threading.Lock()
7

8
    def __new__(cls):
9
        if cls._instance is None:
10
            with cls._lock:
11
                if cls._instance is None:
12
                    cls._instance = super(CacheManager, cls).__new__(cls)
13
                    cls._instance.initialize_cache()
14
        return cls._instance
15

16
    def initialize_cache(self):
17
        self.cache_data = {}  # 实际缓存数据的数据结构
18

19
    def get_data(self, key):
20
        return self.cache_data.get(key)
21

22
    def set_data(self, key, value):
23
        with self._lock:
24
            self.cache_data[key] = value
25

26

27
if __name__ == "__main__":
28
    cache_manager1 = CacheManager()
29
    cache_manager1.set_data("key1", "value1")
30

31
    cache_manager2 = CacheManager()
32
    value = cache_manager2.get_data("key1")
33
    print(value)  # 输出 "value1"
34

35
    print(cache_manager1 is cache_manager2)  # 如果为 True，则是单例模式

【5】线程池#

确保在应用程序中只存在一个线程池的实例，以管理并发任务的执行。

1
import threading
2
from concurrent.futures import ThreadPoolExecutor
3

4

5
class ThreadPoolManager:
6
    _instance = None
7
    _lock = threading.Lock()
8

9
    def __new__(cls):
10
        if cls._instance is None:
11
            with cls._lock:
12
                if cls._instance is None:
13
                    cls._instance = super(ThreadPoolManager, cls).__new__(cls)
14
                    cls._instance.initialize_thread_pool()
15
        return cls._instance
16

17
    def initialize_thread_pool(self):
18
        self.thread_pool = ThreadPoolExecutor(max_workers=4)  # 最大工作线程数
19

20
    def submit_task(self, task_function, *args, **kwargs):
21
        return self.thread_pool.submit(task_function, *args, **kwargs)
22

23

24
if __name__ == "__main__":
25
    thread_pool_manager1 = ThreadPoolManager()
26

27

28
    def sample_task(x):
29
        return x * 2
30

31

32
    future = thread_pool_manager1.submit_task(sample_task, 5)
33
    result = future.result()
34
    print(result)  # 输出 10
35

36
    thread_pool_manager2 = ThreadPoolManager()
37

38
    print(thread_pool_manager1 is thread_pool_manager2)  # 如果为 True，则是单例模式

音乐

音乐

【零】引入#

【1】前言#

【2】为什么要单例模式#

【3】如何实现一个单例#

【4】饿汉式与懒汉式#

（1）饿汉式(着急吃)#

（2）懒汉式（不着急吃）#

【一】类属性#

【1】类产生对象#

【2】类属性包装成方法#

【二】装饰器#

【三】元类(metaclass)#

【四】基于new方法#

【1】思路#

【2】代码#

【3】优势#

【五】基于模块#

【使用场景】#

【1】配置信息#

【2】分布式ID(雪花算法)#

【3】数据库连接池#

【4】缓存管理#

【5】线程池#

支持与分享

音乐

Python中实现单例模式的六种常见方法

【零】引入#

【1】前言#

【2】为什么要单例模式#

【3】如何实现一个单例#

【4】饿汉式与懒汉式#

（1）饿汉式(着急吃)#

（2）懒汉式（不着急吃）#

【一】类属性#

【1】类产生对象#

【2】类属性包装成方法#

【二】装饰器#

【三】元类(metaclass)#

【四】基于__new__方法#

【1】思路#

【2】代码#

【3】优势#

【五】基于模块#

【使用场景】#

【1】配置信息#

【2】分布式ID(雪花算法)#

【3】数据库连接池#

【4】缓存管理#

【5】线程池#

支持与分享

【四】基于new方法#