一、說明
之前寫了“Linux shell腳本單實例模式實現”,python也是日常需要使用的,所以也想來看python中如何實現。
一方面,shell中沒有類和類實例的概念,所以一般說“單實例”都是指“單進程實例”,沒有設計模式中“單例”的概念;另一方面,由於單進程實例和單例都是強調“唯一一份”所以在長時間里以為他們是相同的一個東西,和shell一樣籠統地稱為單實例就好了。
但現在看來他們不是一回事,“單進程實例”討論的環境是整個內存、面向的對象是文件、結果是要么干掉原來的進程新啟一個進程要么結束當前的進程保留原來的進程。
“單例模式”討論的環境是一個進程內、面向的對象是類,結果是不管你在哪、調用多少次返回的都是同一個類實例。也就是說,如果是不同進程,那么是可以返回不同的類實例的(應該說就沒法返回相同的類實例)。
二、單進程實例實現
2.1 Linux平台實現--使用標准庫fcntl
linux平台可以通過python標准庫fcntl來實現鎖
import os import time import fcntl class Test(): # 此函數用於獲取鎖 def _get_lock(self): file_name = os.path.basename(__file__) # 為了統一按linux的習慣放到/var/run目錄去 lock_file_name = f"/var/run/{file_name}.pid" # 是讀還是寫還是什么模式並不重要,因為只是看能不能獲取文件鎖而不一定要寫入內容 # 但是這個一定要是成員變量self.fd而不能是局部變量fd # 因為實際發現當python發現局部變量fd不再使用時會將其回收,這就導致后邊再運行時都能獲取到鎖 self.fd = open(lock_file_name, "w") try: # #define LOCK_SH 1 /* Shared lock. */ 共享鎖 # #define LOCK_EX 2 /* Exclusive lock. */ 互斥鎖 # #define LOCK_UN 8 /* Unlock. */ 解鎖 # LOCK_NB--非阻塞模式。 # 阻塞模式--獲取不到鎖時一直等待 # 非阻塞模式--獲取不到鎖,直接拋出異常 fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB) # 將當前進程號寫入文件 # 如果獲取不到鎖上一步就已經異常了,所以不用擔心覆蓋 self.fd.writelines(str(os.getpid())) # 寫入的數據太少,默認會先被放在緩沖區,我們強制同步寫入到文件 self.fd.flush() except: print(f"{file_name} have another instance running.") exit(1) def __init__(self): self._get_lock() def hello_world(self): print("hello world!") time.sleep(30) # 從觀察到的現像看,占用鎖的進程被關閉后,鎖也就自動釋放了 # 也就是說,其實並不需要在最后自己主動釋放鎖 def __del__(self): fcntl.flock(self.fd, fcntl.LOCK_UN) if __name__ == "__main__": obj = Test() obj.hello_world()
2.2 通用平台實現--使用第三方庫portalocker
安裝方法:pip install portalocker
pypi地址:https://pypi.org/project/portalocker/
github地址:https://github.com/WoLpH/portalocker
import os import time import portalocker class Test(): def _get_lock(self): file_name = os.path.basename(__file__) # linux等平台依然使用標准的/var/run,其他nt等平台使用當前目錄 if os.name == "posix": lock_file_name = f"/var/run/{file_name}.pid" else: lock_file_name = f"{file_name}.pid" self.fd = open(lock_file_name, "w") try: portalocker.lock(self.fd, portalocker.LOCK_EX | portalocker.LOCK_NB) # 將當前進程號寫入文件 # 如果獲取不到鎖上一步就已經異常了,所以不用擔心覆蓋 self.fd.writelines(str(os.getpid())) # 寫入的數據太少,默認會先被放在緩沖區,我們強制同步寫入到文件 self.fd.flush() except: print(f"{file_name} have another instance running.") exit(1) def __init__(self): self._get_lock() def hello_world(self): print("hello world!") time.sleep(30) # 和fcntl有點區別,portalocker釋放鎖直接有unlock()方法 # 還是一樣,其實並不需要在最后自己主動釋放鎖 def __del__(self): portalocker.unlock(self.fd) if __name__ == "__main__": obj = Test() obj.hello_world()
三、單例模式實現
3.1 單例模式示例代碼
import time import threading import datetime class Singleton: _instance_lock = threading.Lock() def __init__(self): pass def __new__(cls, *args, **kwargs): if not hasattr(Singleton, "_instance"): with Singleton._instance_lock: if not hasattr(Singleton, "_instance"): Singleton._instance = super(Singleton, cls).__new__(cls, *args, **kwargs) return Singleton._instance def main_logic(self): # 打印自己及當前時間 print(f"instance--{self}\n" f"now time--{datetime.datetime.now().strftime('%H:%M:%S')}") time.sleep(10) if __name__ == "__main__": obj1 = Singleton() obj2 = Singleton() obj1.main_logic() obj2.main_logic()
3.2 確認單例模式不管實例化多少次都返回同一個對象
運行代碼,可以看到兩個實例是一樣的
3.3 確認單例模式可以有多個進程實例
我們在最開始說單進程實例和單例模式是不同層次的兩個東西,不能相互代替。為了消除這個疑慮,尤其是單例模式可以代替單進程實例的疑慮,我們來做一下實驗。
在相同時間段內,打開兩個窗口分別運行代碼,可以看到兩次都成功了,即使用單例模式的代碼在內存中是可以有多個進程實例的。
參考:
https://stackoverflow.com/questions/28470246/python-lockf-and-flock-behaviour
https://cloud.tencent.com/developer/article/1115821
https://zhuanlan.zhihu.com/p/25134841
https://www.oreilly.com/library/view/python-cookbook/0596001673/ch04s25.html
https://stackoverflow.com/questions/1422368/fcntl-substitute-on-windows