有没有Pythonic的方法来确保只有一个程序实例在运行?
我想到的唯一合理的解决方案是尝试将其作为服务器在某个端口上运行,然后第二个程序尝试绑定到相同的端口-失败。但这不是一个很好的主意,也许有比这更轻量级的东西吗?
(考虑到程序有时会出现故障,例如段错误-因此像“锁定文件”这样的东西将不起作用)
有没有Pythonic的方法来确保只有一个程序实例在运行?
我想到的唯一合理的解决方案是尝试将其作为服务器在某个端口上运行,然后第二个程序尝试绑定到相同的端口-失败。但这不是一个很好的主意,也许有比这更轻量级的东西吗?
(考虑到程序有时会出现故障,例如段错误-因此像“锁定文件”这样的东西将不起作用)
以下代码应该可以完成工作,它是跨平台的,并且可在Python 2.4-3.2上运行。我已在Windows、OS X和Linux上进行过测试。
from tendo import singleton
me = singleton.SingleInstance() # will sys.exit(-1) if other instance is running
最新的代码版本可以在singleton.py中获取。如果发现问题,请在此报告错误。
您可以使用以下方法之一安装tend:
easy_install tendo
pip install tendo
在zgoda的another question中发现的简单的跨平台解决方案:
import fcntl
import os
import sys
def instance_already_running(label="default"):
"""
Detect if an an instance with the label is already running, globally
at the operating system level.
Using `os.open` ensures that the file pointer won't be closed
by Python's garbage collector after the function's scope is exited.
The lock will be released when the program exits, or could be
released if the file pointer were closed.
"""
lock_file_pointer = os.open(f"/tmp/instance_{label}.lock", os.O_WRONLY)
try:
fcntl.lockf(lock_file_pointer, fcntl.LOCK_EX | fcntl.LOCK_NB)
already_running = False
except IOError:
already_running = True
return already_running
这很像S.Lott的建议,但有代码。
fcntl
模块(尽管可以模拟其功能)。 - jfslock_file_pointer = os.open(lock_path, os.O_WRONLY | os.O_CREAT)
。 - baziorek这段代码只适用于Linux系统。它使用了“抽象”的UNIX域套接字,但很简单且不会留下陈旧的锁文件。我喜欢它胜过上面的解决方案,因为它不需要特殊保留的TCP端口。
try:
import socket
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
## Create an abstract socket, by prefixing it with null.
s.bind( '\0postconnect_gateway_notify_lock')
except socket.error as e:
error_code = e.args[0]
error_string = e.args[1]
print "Process already running (%d:%s ). Exiting" % ( error_code, error_string)
sys.exit (0)
唯一的字符串postconnect_gateway_notify_lock
可以更改以允许需要强制单个实例的多个程序。
我不知道这是否足够符合Python的语言风格,但在Java世界中,监听一个已定义的端口是一种广泛使用的解决方案,因为它适用于所有主要平台,并且不会出现崩溃程序的问题。
监听端口的另一个优点是,您可以向正在运行的实例发送命令。例如,当用户第二次启动程序时,您可以向正在运行的实例发送一个命令,告诉它打开另一个窗口(例如Firefox就是这样做的。虽然我不知道它们是否使用TCP端口或命名管道之类的东西)。
import socket; s = socket.socket(socket.AF_INET, socket.SOCK_STREAM); s.bind(('localhost', DEFINED_PORT))
。如果另一个进程绑定到相同的端口,将引发 OSError
。 - crishoj我以前没写过Python,但这是我在mycheckpoint中实现的防止它被crond启动两次或更多次的代码:
import os
import sys
import fcntl
fh=0
def run_once():
global fh
fh=open(os.path.realpath(__file__),'r')
try:
fcntl.flock(fh,fcntl.LOCK_EX|fcntl.LOCK_NB)
except:
os._exit(0)
run_once()
在另一个问题(http://stackoverflow.com/questions/2959474)中发布后,发现了Slava-N的建议。这个建议被称为一个函数,可以锁定执行脚本的文件(而不是pid文件),并在脚本结束(正常或错误)之前保持锁定状态。
使用 pid 文件。你有一个已知的位置“/path/to/pidfile”,在启动时,您可以执行类似以下代码(部分伪代码因为我还没喝咖啡,不想太努力):
import os, os.path
pidfilePath = """/path/to/pidfile"""
if os.path.exists(pidfilePath):
pidfile = open(pidfilePath,"r")
pidString = pidfile.read()
if <pidString is equal to os.getpid()>:
# something is real weird
Sys.exit(BADCODE)
else:
<use ps or pidof to see if the process with pid pidString is still running>
if <process with pid == 'pidString' is still running>:
Sys.exit(ALREADAYRUNNING)
else:
# the previous server must have crashed
<log server had crashed>
<reopen pidfilePath for writing>
pidfile.write(os.getpid())
else:
<open pidfilePath for writing>
pidfile.write(os.getpid())
换句话说,你正在检查pid文件是否存在;如果不存在,则将你的pid写入该文件。 如果pid文件存在,则检查该pid是否是运行中进程的pid;如果是,则表示有另一个正在运行的进程,因此只需关闭即可。 如果不是,则先记录之前的进程已崩溃,然后用自己的pid替换原来的pid并继续执行。
import win32event
import win32api
from winerror import ERROR_ALREADY_EXISTS
mutex = win32event.CreateMutex(None, False, 'name')
last_error = win32api.GetLastError()
if last_error == ERROR_ALREADY_EXISTS:
print("App instance already running")
有些答案使用fctnl
(也包含在@sorin tendo包中),但它在Windows上不可用。如果您尝试使用像pyinstaller
这样的软件包冻结Python应用程序并进行静态导入,则会引发错误。
此外,使用锁定文件方法会导致数据库文件出现只读
问题(我在使用sqlite3
时遇到过)。
以下是我最终的Windows-only解决方案。将以下内容放入一个模块中,可以取名为'onlyone.py'或其他名称。直接在您的__main__ Python脚本文件中包含该模块即可。
import win32event, win32api, winerror, time, sys, os
main_path = os.path.abspath(sys.modules['__main__'].__file__).replace("\\", "/")
first = True
while True:
mutex = win32event.CreateMutex(None, False, main_path + "_{<paste YOUR GUID HERE>}")
if win32api.GetLastError() == 0:
break
win32api.CloseHandle(mutex)
if first:
print "Another instance of %s running, please wait for completion" % main_path
first = False
time.sleep(1)
代码尝试创建一个互斥锁,其名称从脚本的完整路径派生而来。我们使用正斜杠来避免与实际文件系统产生混淆。
对于任何使用wxPython开发应用程序的人,您可以使用函数 wx.SingleInstanceChecker
在这里记录。
我个人使用wx.App
的子类,它使用wx.SingleInstanceChecker
并从OnInit()
返回False
,如果已经存在正在执行的应用程序实例,则会像这样:
import wx
class SingleApp(wx.App):
"""
class that extends wx.App and only permits a single running instance.
"""
def OnInit(self):
"""
wx.App init function that returns False if the app is already running.
"""
self.name = "SingleApp-%s".format(wx.GetUserId())
self.instance = wx.SingleInstanceChecker(self.name)
if self.instance.IsAnotherRunning():
wx.MessageBox(
"An instance of the application is already running",
"Error",
wx.OK | wx.ICON_WARNING
)
return False
return True
wx.App
并禁止多个实例。使用它只需在代码中将 wx.App
替换为 SingleApp
即可:app = SingleApp(redirect=False)
frame = wx.Frame(None, wx.ID_ANY, "Hello World")
frame.Show(True)
app.MainLoop()
这可能有效。
尝试在已知位置创建PID文件。如果失败,则有人锁定了该文件,您就完成了。
正常完成后,关闭并删除PID文件,以便其他人可以覆盖它。
您可以将程序包装在shell脚本中,即使程序崩溃也可以删除PID文件。
您还可以使用PID文件来杀死挂起的程序。