Python脚本如何实现单实例运行

wen python案例 28

本文目录导读:

Python脚本如何实现单实例运行

  1. 使用锁文件(最常用)
  2. 使用系统互斥量(Windows)
  3. 使用端口占用(跨平台简单方案)
  4. 使用第三方库
  5. 完整的跨平台解决方案
  6. 推荐使用方案

我来介绍几种在Python中实现单实例运行的方法:

使用锁文件(最常用)

文件锁(跨平台)

import os
import sys
import fcntl
class SingleInstance:
    def __init__(self, lock_file='/tmp/my_app.lock'):
        self.lock_file = lock_file
        self.fp = None
    def __enter__(self):
        try:
            self.fp = open(self.lock_file, 'w')
            fcntl.flock(self.fp.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except (IOError, OSError):
            print("程序已在运行中!")
            sys.exit(1)
        return self
    def __exit__(self, *args):
        if self.fp:
            fcntl.flock(self.fp.fileno(), fcntl.LOCK_UN)
            self.fp.close()
            try:
                os.unlink(self.lock_file)
            except:
                pass
# 使用示例
if __name__ == '__main__':
    with SingleInstance():
        # 这里是你的主程序逻辑
        import time
        print("程序开始运行...")
        time.sleep(10)
        print("程序结束")

简化版文件锁

import os
import sys
import atexit
def single_instance_check(lock_file='/tmp/myapp.lock'):
    """检查并确保单实例运行"""
    try:
        # 尝试创建锁文件
        fd = os.open(lock_file, os.O_CREAT | os.O_EXCL | os.O_RDWR)
        # 注册退出时删除锁文件
        def cleanup():
            os.close(fd)
            os.unlink(lock_file)
        atexit.register(cleanup)
        return True
    except FileExistsError:
        return False
# 使用示例
if __name__ == '__main__':
    if not single_instance_check():
        print("程序已在运行中!")
        sys.exit(1)
    # 主程序逻辑
    import time
    print("程序开始运行...")
    time.sleep(10)

使用系统互斥量(Windows)

import sys
import win32event
import win32api
import winerror
def single_instance_win(mutex_name='MyAppMutex'):
    """Windows系统使用互斥量实现单实例"""
    try:
        mutex = win32event.CreateMutex(None, False, mutex_name)
        if win32api.GetLastError() == winerror.ERROR_ALREADY_EXISTS:
            return False
        return True
    except:
        return False
# 使用示例
if __name__ == '__main__':
    if not single_instance_win():
        print("程序已在运行中!")
        sys.exit(1)
    # 主程序逻辑
    print("程序开始运行...")

使用端口占用(跨平台简单方案)

import socket
import sys
def single_instance_port(port=12345):
    """通过占用端口实现单实例"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        sock.bind(('localhost', port))
        return sock  # 返回socket对象,保持端口占用
    except socket.error:
        return None
# 使用示例
if __name__ == '__main__':
    sock = single_instance_port()
    if not sock:
        print("程序已在运行中!")
        sys.exit(1)
    # 主程序逻辑
    import time
    print("程序开始运行...")
    try:
        time.sleep(10)
    finally:
        sock.close()

使用第三方库

pip install tendo
from tendo import singleton
import sys
def main():
    try:
        me = singleton.SingleInstance()
        # 主程序逻辑
        print("程序开始运行...")
        import time
        time.sleep(10)
    except singleton.SingleInstanceException:
        print("程序已在运行中!")
        sys.exit(1)
if __name__ == '__main__':
    main()

完整的跨平台解决方案

import os
import sys
import tempfile
import platform
class SingleInstance:
    """跨平台单实例实现"""
    def __init__(self, name='myapp'):
        self.name = name
        self.fp = None
        self.lock_file = None
    def run(self, target_func, *args, **kwargs):
        if platform.system() == 'Windows':
            return self._run_windows(target_func, *args, **kwargs)
        else:
            return self._run_unix(target_func, *args, **kwargs)
    def _run_unix(self, target_func, *args, **kwargs):
        import fcntl
        self.lock_file = os.path.join(tempfile.gettempdir(), f'{self.name}.lock')
        try:
            self.fp = open(self.lock_file, 'w')
            fcntl.flock(self.fp.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            target_func(*args, **kwargs)
            return True
        except (IOError, OSError):
            print(f"程序 '{self.name}' 已在运行中!")
            return False
        finally:
            if self.fp:
                fcntl.flock(self.fp.fileno(), fcntl.LOCK_UN)
                self.fp.close()
                try:
                    os.unlink(self.lock_file)
                except:
                    pass
    def _run_windows(self, target_func, *args, **kwargs):
        import win32event
        import win32api
        import winerror
        mutex = win32event.CreateMutex(None, False, f'Global\\{self.name}')
        if win32api.GetLastError() == winerror.ERROR_ALREADY_EXISTS:
            print(f"程序 '{self.name}' 已在运行中!")
            return False
        target_func(*args, **kwargs)
        win32api.CloseHandle(mutex)
        return True
# 使用示例
def my_app():
    import time
    print("主程序开始运行...")
    time.sleep(10)
    print("主程序结束")
if __name__ == '__main__':
    app = SingleInstance('my_python_app')
    app.run(my_app)

推荐使用方案

  1. 简单场景:使用 文件锁 方案(方法一)
  2. Windows只:使用 互斥量 方案
  3. 跨平台:使用 tendo 库或自定义跨平台方案
  4. 服务类应用:使用 端口占用 方案

选择哪种方案取决于你的具体需求和运行环境。

抱歉,评论功能暂时关闭!