python 单一进程实例实现

左手的ㄟ右手 2021-09-02 03:12 410阅读 0赞

**目录**

一、说明

二、单进程实例实现

2.1 Linux平台实现--使用标准库fcntl

2.2 通用平台实现--使用第三方库portalocker

三、单例模式实现

3.1 单例模式示例代码

3.2 确认单例模式不管实例化多少次都返回同一个对象

3.3 确认单例模式可以有多个进程实例

--------------------

## 一、说明 ##

之前写了“[Linux shell脚本单实例模式实现][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/][https_pypi.org_project_portalocker]

github地址：[https://github.com/WoLpH/portalocker][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 确认单例模式不管实例化多少次都返回同一个对象 ###

运行代码，可以看到两个实例是一样的

![f11a05a6c45f33ed6281e552cdb908cf.png][]

### 3.3 确认单例模式可以有多个进程实例 ###

我们在最开始说单进程实例和单例模式是不同层次的两个东西，不能相互代替。为了消除这个疑虑，尤其是单例模式可以代替单进程实例的疑虑，我们来做一下实验。

在相同时间段内，打开两个窗口分别运行代码，可以看到两次都成功了，即使用单例模式的代码在内存中是可以有多个进程实例的。

![d33703c4003f7e47ae184bb2cd8609ea.png][]

![a82db1ff222bace5a1b4b5955d4bbaf1.png][]

[Linux shell]: https://www.cnblogs.com/lsdb/p/12010128.html
[https_pypi.org_project_portalocker]: https://pypi.org/project/portalocker/
[https_github.com_WoLpH_portalocker]: https://github.com/WoLpH/portalocker
[f11a05a6c45f33ed6281e552cdb908cf.png]: /images/20210728/2c0691502ab6482189843462a952c05f.png
[d33703c4003f7e47ae184bb2cd8609ea.png]: /images/20210728/15f8f92967d34a87bdcc7ada198a2c39.png
[a82db1ff222bace5a1b4b5955d4bbaf1.png]: /images/20210728/58792368ab304dd1a6c72ce9c79b7b10.png