Python:程序的单一实例

有没有Pythonic方法只有一个程序的一个实例运行?

我提出的唯一合理的解决scheme是试图运行它作为一个端口上的服务器,然后第二个程序试图绑定到相同的端口 – 失败。 但这不是一个好主意,也许有比这更轻量级的东西?

(考虑到程序有时会失败,即segfault – 所以像“locking文件”的东西不会工作)

更新 :提供的解决scheme比只有一个不存在的服务器占用一个端口更加复杂和不太依赖,所以我必须去那个。

下面的代码应该做的工作,它是跨平台,并运行在Python 2.4-3.2。 我在Windows,OS X和Linux上testing了它。

from tendo import singleton me = singleton.SingleInstance() # will sys.exit(-1) if other instance is running 

最新的代码版本是singleton.py 。 请在这里提交错误 。

您可以使用以下方法之一安装趋势:

简单的跨平台解决scheme,在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) 

很像S.洛特的build议,但与代码。

这个代码是Linux专用的(它使用“抽象的”UNIX域套接字),但它很简单,不会留下过时的locking文件。 我更喜欢上面的解决scheme,因为它不需要特别保留的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) 

唯一的stringpostconnect_gateway_notify_lock可以更改为允许需要单个实例的多个程序强制执行。

我不知道pythonic是否足够,但在Java世界里,在定义的端口上进行侦听是一个非常广泛使用的解决scheme,因为它可以在所有主要平台上运行,并且在程序崩溃时不会有任何问题。

监听端口的另一个好处是你可以发送一个命令给正在运行的实例。 例如,当用户第二次启动程序时,你可以发送运行实例的一个命令来告诉它打开另一个窗口(这就是Firefox所做的),我不知道他们是使用TCP端口还是命名pipe道,或者类似的东西,'虽然)。

以前从来没有写过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() 

在另一个问题发布之后发现Slava-N的build议(http://stackoverflow.com/questions/2959474)。; 这个被称为函数,locking正在执行的脚本文件(不是一个pid文件),并维护locking,直到脚本结束(正常或错误)。

使用一个pid文件。 你有一些已知的位置,“/path/到/ 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到该文件。 如果pidfile确实存在,则检查pid是否是正在运行的进程的pid; 如果是这样的话,那么你又有一个正在运行的进程,所以就closures了。 如果没有,那么之前的进程崩溃,所以logging下来,然后写你自己的PID到文件,而不是旧的。 然后继续。

您已经在另一个线程中find类似问题的回复,为了完整起见,请参阅如何在Windows上实现相同的命名互斥锁。

http://code.activestate.com/recipes/474070/

这可能工作。

  1. 尝试创build一个PID文件到已知的位置。 如果你失败了,有人将文件locking,你就完成了。

  2. 当你正常完成,closures并删除PID文件,所以别人可以覆盖它。

即使您的程序崩溃,您也可以将程序封装在一个shell脚本中,以便删除PID文件。

如果程序挂起,也可以使用PID文件来终止程序。

使用locking文件是unix上非常常见的一种方法。 如果它崩溃,你必须手动清理。 您可以在文件中存储PID,并在启动时检查是否存在具有此PID的进程,如果不是,则覆盖locking文件。 (但是,您还需要lockingread-file-check-pid-rewrite-file)。 你会发现你需要获得和检查os包中的 pid。 检查是否存在具有给定PID的进程的常见方式是发送一个非致命的信号。

其他的select可能是将这个与群或posix信号灯相结合。

按照Saua的build议,打开networking套接字可能是最简单和最便携的。

对于任何使用wxPython作为应用程序的人,可以使用这里logging的函数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 ,禁止多个实例。 要使用它,只需用代码中的SingleAppreplacewx.App ,如下所示:

 app = SingleApp(redirect=False) frame = wx.Frame(None, wx.ID_ANY, "Hello World") frame.Show(True) app.MainLoop() 

我发布这个答案,因为我是一个新的用户和堆栈溢出不会让我投票呢。

Sorin Sbarnea的解决scheme在OS X,Linux和Windows下适用于我,我很感激。

但是,tempfile.gettempdir()在OS X和Windows下运行,另一个在其他一些/ many / all(?)* nixes下(忽略OS X也是Unix!)。 这个区别对于这个代码很重要。

OS X和Windows具有用户特定的临时目录,因此一个用户创build的临时文件对另一个用户不可见。 相比之下,在许多版本的* nix(我testing过Ubuntu 9,RHEL 5,OpenSolaris 2008和FreeBSD 8)中,临时目录是/ tmp,适用于所有用户。

这意味着在多用户机器上创buildlocking文件时,它将在/ tmp中创build,并且只有第一次创buildlocking文件的用户才能运行该应用程序。

一个可能的解决scheme是将当前用户名embedded到locking文件的名称中。

值得注意的是,OP的抓取端口的解决scheme也会在多用户机器上出现问题。

我在gentoo上使用single_process ;

 pip install single_process 

例如

 from single_process import single_process @single_process def main(): print 1 if __name__ == "__main__": main() 

请参阅: https : //pypi.python.org/pypi/single_process/1.0

linux的例子

此方法基于在closures应用程序后自动删除的临时文件的创build。 程序启动,我们validation文件的存在; 如果文件存在(有待执行),程序closures; 否则会创build文件并继续执行程序。

 from tempfile import * import time import os import sys f = NamedTemporaryFile( prefix='lock01_', delete=True) if not [f for f in os.listdir('/tmp') if f.find('lock01_')!=-1] else sys.exit() YOUR CODE COMES HERE 

我一直怀疑应该是一个很好的使用进程组的POSIXy解决scheme,而不必击中文件系统,但是我不能完全明白。 就像是:

启动时,您的进程向特定组中的所有进程发送“kill -0”。 如果存在这样的过程,则退出。 然后它join组。 没有其他进程使用该组。

然而,这有一个竞争条件 – 多个进程都可以在同一时间完成这一切,并最终join该组并同时运行。 当你添加某种互斥体来使其水密,你不再需要进程组。

这可能是可以接受的,如果你的过程只能由cron开始,每分钟一次或每小时一次,但是让我有点紧张,就是在你不想要的时候会出错。

我想这毕竟不是一个很好的解决办法,除非有人能改进呢?

上周我遇到了这个问题,尽pipe我find了一些很好的解决scheme,但我决定做一个非常简单,干净的python包,并将其上传到PyPI。 它与tendo不同,它可以locking任何string资源名称。 虽然你可以locking__file__来达到同样的效果。

安装: pip install quicklock

使用它非常简单:

 [nate@Nates-MacBook-Pro-3 ~/live] python Python 2.7.6 (default, Sep 9 2014, 15:04:36) [GCC 4.2.1 Compatible Apple LLVM 6.0 (clang-600.0.39)] on darwin Type "help", "copyright", "credits" or "license" for more information. >>> from quicklock import singleton >>> # Let's create a lock so that only one instance of a script will run ... >>> singleton('hello world') >>> >>> # Let's try to do that again, this should fail ... >>> singleton('hello world') Traceback (most recent call last): File "<stdin>", line 1, in <module> File "/Users/nate/live/gallery/env/lib/python2.7/site-packages/quicklock/quicklock.py", line 47, in singleton raise RuntimeError('Resource <{}> is currently locked by <Process {}: "{}">'.format(resource, other_process.pid, other_process.name())) RuntimeError: Resource <hello world> is currently locked by <Process 24801: "python"> >>> >>> # But if we quit this process, we release the lock automatically ... >>> ^D [nate@Nates-MacBook-Pro-3 ~/live] python Python 2.7.6 (default, Sep 9 2014, 15:04:36) [GCC 4.2.1 Compatible Apple LLVM 6.0 (clang-600.0.39)] on darwin Type "help", "copyright", "credits" or "license" for more information. >>> from quicklock import singleton >>> singleton('hello world') >>> >>> # No exception was thrown, we own 'hello world'! 

看看: https : //pypi.python.org/pypi/quicklock

这是我最终的纯Windows解决scheme。 把以下内容放到一个模块中,或许叫做'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) 

说明

代码尝试创build一个名为派生自脚本完整path的名称的互斥锁。 我们使用正斜杠来避免与真实文件系统的混淆。

优点

  • 不需要configuration或“魔术”标识符,根据需要以多种不同的脚本使用它。
  • 没有陈旧的文件留下来,互斥与你一起死亡。
  • 等待时打印一条有用的信息

我只会做一些事情:

 if commands.getstatusoutput("mkdir /tmp/test")[0]: print "Exiting" sys.exit(1) 

在我的代码的最后,我将删除该目录。 mkdir是一个primefaces,在这里工作得很好。