【centos】【python】程序单例

需求：

避免同一个程序在不同时刻、终端并行执行；不是绝对不能，是两个同样的程序尽量不要一块执行；

技术：

在程序执行时将pid写入到一个文件里，结束时再删除pid文件，下一个程序先检查pid文件是否存在再决定是否提前终止；

import datetimeimport shutilfrom pathlib import Pathimport loggingimport os# import fcntlimport psutilfrom logging.handlers import RotatingFileHandlerLOG_PATH_PARENT = "./"LOG_PATH = LOG_PATH_PARENT + "/log"logger = logging.getLogger()logger.setLevel(logging.INFO)formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s')log_abs_file = os.path.join(LOG_PATH_PARENT, 'suri_logrotate.log')handler = RotatingFileHandler(log_abs_file, maxBytes=1024 * 1024 * 10, backupCount=10)# 大小10MB , 10个handler.setFormatter(formatter)logger.addHandler(handler)console = logging.StreamHandler()console.setFormatter(formatter)logger.addHandler(console)class SingletonProgress:def _get_lock(self):"""开启进程占用锁，确保其他同名称进程不再重复运行"""# file_name = os.path.basename(__file__)# lock_file_name = f"/var/run/{file_name}.pid"# self.fd = open(self.lock_file_name, "w")if not os.path.exists(self.lock_file_name):with open(self.lock_file_name, "w", encoding='utf-8') as f:f.write(str(os.getpid()))else:if os.path.exists(self.lock_file_name):print(self.lock_file_name)with open(self.lock_file_name, 'r', encoding='utf8') as f:pid = f.read().strip()if not pid:logger.error("Program abnormal exit,no pid, Please try again.")os.remove(self.lock_file_name)elif pid and pid.isdigit() and int(pid) not in psutil.pids():logger.error("Program abnormal exit,pid err, Please try again.")os.remove(self.lock_file_name)elif self.file_name not in psutil.Process(int(pid)).cmdline():logger.error("Program abnormal exit,pid not, Please try again.")os.remove(self.lock_file_name)else:logger.info(f"{self.file_name} have another instance running.")self.remove_lock_file = Falseexit(1)def __init__(self):self.file_name = os.path.basename(__file__)self.lock_file_name = os.path.join(LOG_PATH_PARENT, self.file_name+'.pid')self.remove_lock_file = Trueself.err_delete_lock_file_name = Falseself._get_lock()def __del__(self):# 占用锁的进程被关闭后，释放锁if self.remove_lock_file:os.remove(self.lock_file_name)def method():import timewhile 1:time.sleep(1)logger.info(str(os.getpid()))if __name__ == '__main__':keep_lock = SingletonProgress()method()# del keep_lock