多線程環境下實現單例模式

下面介紹了兩種Python實現單例模式的方法

1.重寫__new__方法實現多線程情況下的單例模式

用new方法實現單例模式

import time, threadingclass Singleton:"""單例模式————最多只允許創建一個該類實例"""_lock = threading.Lock()_instance = Nonedef __new__(cls, a, b, *args, **kwargs):time.sleep(1) # 模擬高并發情況with cls._lock:# print("進入lock區域")if not cls._instance:cls._instance = object.__new__(cls)# cls._instance = super().__new__(cls) # 與上一條作用相同# cls._instance.a = areturn cls._instance# S = Singleton(1, 2) # print(S)def task(a, b):s = Singleton(a, b)print(s)for i in range(5):t = threading.Thread(target=task, args=(2, 3))t.start()

運行始終為單一實例：

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

2.使用裝飾器實現多線程情況下的單例模式

和前面一種方法不一樣，沒有使用類的內部屬性記錄實例對象，而是使用了字典鍵唯一的特性，將類對象Example作為鍵，實例化后的對象作為值
錯誤示例：

import time from functools import wraps import threadingdef singleton_dec(class_):map_ = {} # 定義一個字典，類本身作為鍵，實例化后的對象作為其相應的值@wraps(class_) # （順便復習一下裝飾器，將被裝飾的函數改回原來的屬性和名字）def singleton(*args):# print(class_.__name__)# print(class_.foo.__name__)if class_ not in map_: # 如果類沒有在字典就實例化time.sleep(1) # 沒有加鎖的情況下模擬多進程的情況map_[class_] = class_(*args) # 實例化類方法并存到字典中return map_[class_] # 字典中有這個類就不實例化return singleton@singleton_dec class EXAMPLE:def __init__(self, param='abc'):self.param = paramdef foo(self):pass# E1 = EXAMPLE('123') # print(E1.param) # print(E1) # E2 = EXAMPLE('456') # 由于EXAMPLE已經實例化了（在map_中），這一步并沒有對EXAMPLE做出任何改變 # print(E2.param) # 輸出的依然是123 # print(E2)def task():e = EXAMPLE()print(e)for i in range(5):t = threading.Thread(target=task)t.start()

加上sleep模擬多線程的執行結果：

<__main__.EXAMPLE object at 0x000001BD894A7128> <__main__.EXAMPLE object at 0x000001BD8949E710> <__main__.EXAMPLE object at 0x000001BD894A1DD8> <__main__.EXAMPLE object at 0x000001BD8949EB00> <__main__.EXAMPLE object at 0x000001BD8949E710>

可以看到在并發情況很高的時候并沒有實現單例化

因為方法二沒有像方法一一樣設置線程鎖，所以這里會出現創建多個實例的情況

想要實現單例，也需要像方法一中一樣加上線程鎖

不過是給字典方法加上一個線程鎖：
示例：

# -*- coding: utf-8 -*- import threading, time from functools import wrapsclass DictWithLock(dict):lock = threading.Lock()def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)def singleton_decorator(callable_obj):dl = DictWithLock()@wraps(callable_obj)def inner(*args, **kwargs):# print('測試', *args, **kwargs)time.sleep(1)with dl.lock: # 一次只能有一個線程/進程使用DictWithLock的lock屬性# print("進入lock區域")if not dl.get("_instance"):dl["_instance"] = callable_obj(*args, **kwargs)return dl["_instance"]return inner@singleton_decorator class Test:def __init__(self, a=0, b=1):print('Test.test', a, b)def task(a, b):t = Test(a, b)print(t)# t0 = Test(2, 3) # print(t0)for i in range(5):tsk = threading.Thread(target=task, args=(2, 3))tsk.start()

輸出結果：

Test.test 2 3 <__main__.Test object at 0x00000292250AB070> <__main__.Test object at 0x00000292250AB070><__main__.Test object at 0x00000292250AB070><__main__.Test object at 0x00000292250AB070><__main__.Test object at 0x00000292250AB070>進程已結束，退出代碼 0

線程各自異步執行，但始終只會創建一個Example的實例

限制實例個數的模式————limit實例模式

__new__()方法實現最多創建3個實例的模式：

class LimitedInstances:"""限制實例個數的模式————limit實例模式"""_instances = []limit = 3def __new__(cls, *args, **kwargs):if not len(cls._instances) < cls.limit: # 若個數達到limit說明已滿raise Exception(f"Can not create instance more than {cls.limit}.")instance = super().__new__(cls)cls._instances.append(instance)return instancedef __delete__(self):self._instances.remove(self)LI1 = LimitedInstances(1, 2) LI2 = LimitedInstances(2, 2)LI3 = LimitedInstances(3, 2) LI3.__delete__() LI4 = LimitedInstances(4, 2) print(f"刪除一個又加進來一個，還是3個\n", LimitedInstances._instances) # LI5 = LimitedInstances(5, 2) # 超過3個會報錯

運行結果，限制最多3個實例：

刪除一個又加進來一個，還是3個實例[<__main__.LimitedInstances object at 0x00000278A03BDE20>, <__main__.LimitedInstances object at 0x00000278A03BDDC0>, <__main__.LimitedInstances object at 0x00000278A03BD520>]

模擬高并發情況下，還是可能會出現超過3個實例的情況：

import time, threadingclass LimitedInstances:"""限制實例個數的模式————limit實例模式"""_instances = []limit = 3def __new__(cls, *args, **kwargs):if not len(cls._instances) < cls.limit: # 若個數達到limit說明已滿raise Exception(f"Can not create instance more than {cls.limit}.")instance = super().__new__(cls)time.sleep(1) # 模擬高并發情況cls._instances.append(instance)return instancedef __delete__(self):self._instances.remove(self)def task():LI = LimitedInstances()print(LI)def task():LI = LimitedInstances()print(LI)for i in range(6):t = threading.Thread(target=task, args=())t.start() time.sleep(2) print(len(LimitedInstances._instances), LimitedInstances._instances)

運行結果，出現了6個不同的實例的情況：

<__main__.LimitedInstances object at 0x0000027066EC8DF0><__main__.LimitedInstances object at 0x0000027066EED370><__main__.LimitedInstances object at 0x0000027066EC8BB0> <__main__.LimitedInstances object at 0x0000027066EED0D0> <__main__.LimitedInstances object at 0x0000027066EC8610> <__main__.LimitedInstances object at 0x0000027066EED610> 6 [<__main__.LimitedInstances object at 0x0000027066EC8DF0>, <__main__.LimitedInstances object at 0x0000027066EED370>, <__main__.LimitedInstances object at 0x0000027066EC8BB0>, <__main__.LimitedInstances object at 0x0000027066EED0D0>, <__main__.LimitedInstances object at 0x0000027066EC8610>, <__main__.LimitedInstances object at 0x0000027066EED610>]

解決方式，同之前，加上線程鎖：

import time, threadingclass LimitedInstances:"""限制實例個數的模式————limit實例模式"""_instances = []limit = 3_lock = threading.Lock()def __new__(cls, *args, **kwargs):with cls._lock: # 一次只能一個線程執行實例個數判斷和實例存入if not len(cls._instances) < cls.limit: # 若個數達到limit說明已滿raise Exception(f"Can not create instance more than {cls.limit}.")instance = super().__new__(cls)time.sleep(1) # 模擬高并發情況cls._instances.append(instance)return instancedef __delete__(self):self._instances.remove(self)def task():LI = LimitedInstances()print(LI)for i in range(6):t = threading.Thread(target=task, args=())t.start() # time.sleep(2) # print(len(LimitedInstances._instances), LimitedInstances._instances)

運行時，創建超過了3個實例會報錯

總結

以上是生活随笔為你收集整理的Python多线程下实现单例模式，以及limit实例模式的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。