Sla*_*a V 112 python locking process mutual-exclusion
是否有Pythonic方法只运行一个程序实例?
我提出的唯一合理的解决方案是尝试在某个端口上将其作为服务器运行,然后第二个程序尝试绑定到同一个端口 - 失败.但这不是一个好主意,也许有比这更轻巧的东西?
(考虑到程序有时会失败,即段错误 - 所以像"锁定文件"这样的东西不起作用)
更新:提供的解决方案比仅存在一个不存在的服务器的端口要复杂得多且不太依赖,所以我必须使用那个.
sor*_*rin 91
以下代码应该完成这项工作,它是跨平台的,可以在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
Run Code Online (Sandbox Code Playgroud)
最新的代码版本是singleton.py.请在这里提交错误.
您可以使用以下方法之一安装tend:
easy_install tendopip install tendoSla*_*a V 37
简单,跨平台的解决方案,在发现了另一个问题由Zgoda酒店:
import fcntl, sys
pid_file = 'program.pid'
fp = open(pid_file, 'w')
try:
fcntl.lockf(fp, fcntl.LOCK_EX | fcntl.LOCK_NB)
except IOError:
# another instance is running
sys.exit(0)
Run Code Online (Sandbox Code Playgroud)
很像S.Lott的建议,但有代码.
Rob*_*rio 28
此代码特定于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)
Run Code Online (Sandbox Code Playgroud)
postconnect_gateway_notify_lock可以更改唯一字符串以允许需要强制执行单个实例的多个程序.
Joa*_*uer 23
我不知道它是否足够pythonic,但在Java世界中,侦听已定义的端口是一个非常广泛使用的解决方案,因为它适用于所有主要平台,并且没有任何崩溃程序的问题.
侦听端口的另一个好处是可以向正在运行的实例发送命令.例如,当用户第二次启动程序时,您可以向正在运行的实例发送一个命令,告诉它打开另一个窗口(例如,这就是Firefox所做的.我不知道他们是否使用TCP端口或命名管道或类似的东西,'虽然).
小智 12
从来没有写过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()
Run Code Online (Sandbox Code Playgroud)
在另一期(http://stackoverflow.com/questions/2959474)中发布此事后,发现了Slava-N的建议.这个被称为函数,锁定执行脚本文件(不是pid文件)并保持锁定直到脚本结束(正常或错误).
Cha*_*tin 10
使用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())
Run Code Online (Sandbox Code Playgroud)
所以,换句话说,你正在检查是否存在pid文件; 如果没有,请将您的pid写入该文件.如果pidfile确实存在,那么检查pid是否是正在运行的进程的pid; 如果是这样,那么你有另一个正在运行的进程,所以只需关闭.如果没有,那么前一个进程崩溃,所以记录它,然后将自己的pid写入文件而不是旧文件.然后继续.
您已经在另一个线程中找到了对类似问题的回复,因此为了完整起见,请参阅如何在Windows上实现相同的名为mutex.
http://code.activestate.com/recipes/474070/
这可能有效。
尝试将PID文件创建到已知位置。如果失败,则有人将文件锁定了,您就完成了。
正常完成后,请关闭并删除PID文件,以便其他人可以覆盖它。
您可以将程序包装在Shell脚本中,即使程序崩溃,该脚本也会删除PID文件。
如果程序挂起,也可以使用PID文件将其杀死。
对于任何在其应用程序中使用wxPython 的人,您可以使用wx.SingleInstanceChecker 此处记录的函数 。
我个人使用一个子类,如果存在已经执行的应用程序的现有实例,则wx.App使用该子类wx.SingleInstanceChecker并False从中返回,如下所示:OnInit()
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
Run Code Online (Sandbox Code Playgroud)
wx.App这是对禁止多个实例的简单替代。要使用它,只需在代码中替换为wx.App:SingleApp
app = SingleApp(redirect=False)
frame = wx.Frame(None, wx.ID_ANY, "Hello World")
frame.Show(True)
app.MainLoop()
Run Code Online (Sandbox Code Playgroud)
这是我最终的仅适用于 Windows 的解决方案。将以下内容放入一个模块中,可能称为“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)
Run Code Online (Sandbox Code Playgroud)
该代码尝试创建一个互斥锁,其名称源自脚本的完整路径。我们使用正斜杠来避免与真实文件系统的潜在混淆。
在 Windows 上最好的解决方案是使用 @zgoda 建议的互斥锁。
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")
Run Code Online (Sandbox Code Playgroud)
一些答案使用fctnl(也包含在@sorin tendo 包中)在 Windows 上不可用,如果您尝试使用pyinstaller静态导入之类的包来冻结您的 python 应用程序,它会引发错误。
此外,使用锁定文件方法会导致read-only数据库文件出现问题(遇到过这种情况sqlite3)。