确保程序只运行一个实例

157

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

我想到的唯一合理的解决方案是尝试将其作为服务器在某个端口上运行,然后第二个程序尝试绑定到相同的端口-失败。但这不是一个很好的主意,也许有比这更轻量级的东西吗?

(考虑到程序有时会出现故障,例如段错误-因此像“锁定文件”这样的东西将不起作用)


1
也许如果你追踪并修复了段错误,你的生活会更轻松。但这并不是一件容易的事情。 - David Locke
1
它不在我的库中,而是在Python的libxml绑定中,并且非常腼腆 - 只会隔几天才触发一次。 - Slava V
5
Python标准库支持flock()函数,这对于现代UNIX程序来说是正确的选择。打开端口会占用受限的命名空间中的一个位置,而pid文件更为复杂,因为您需要检查运行进程以安全地使它们失效;而flock()函数则没有这些问题。 - Charles Duffy
这也可以通过命令行实用程序flock在Python之外进行管理。 - Asclepius
25个回答

122

以下代码应该可以完成工作,它是跨平台的,并且可在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:


2
我更新了答案并包含了最新版本的链接。如果你发现了一个 bug,请将其提交到 Github,我会尽快解决它。 - sorin
2
@Johny_M 谢谢,我制作了一个补丁并在http://pypi.python.org/pypi/tendo 上发布了新版本。 - sorin
2
这个语法在 Python 2.6 的 Windows 版本中对我无效。适用于我的是:1:从 tendo 导入 singleton2:me = singleton.SingleInstance() - Brian
42
这么微不足道的东西还需要另一个依赖?听起来不太吸引人。 - WhyNotHugo
3
单例对象处理接收到sigterm信号(例如进程运行时间过长)的进程,还是我需要自己处理? - JimJty
显示剩余11条评论

56

zgodaanother 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的建议,但有代码。


1
Windows系统上没有fcntl模块(尽管可以模拟其功能)。 - jfs
这很不错!比起从外部调用ps和grep等命令要好得多。 - Danylo Gurianov
12
提示:如果你想将此内容封装在函数中,“fp”必须是全局的,否则函数退出后文件将被关闭。 - cmcginty
1
@Mirko Control+Z并不能退出应用程序(在我所知道的任何操作系统上都不行),它只是将其挂起。可以使用“fg”将应用程序返回到前台。因此,听起来它对您来说是正常工作的(即应用程序仍然处于活动状态,但已被挂起,因此锁定仍然存在)。 - Sam Bull
10
在我的情况下(在Linux上的Python 3.8.3),这段代码需要修改: lock_file_pointer = os.open(lock_path, os.O_WRONLY | os.O_CREAT) - baziorek
显示剩余5条评论

42

这段代码只适用于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可以更改以允许需要强制单个实例的多个程序。


1
Roberto,你确定在内核崩溃或硬重置后,文件\0postconnect_gateway_notify_lock不会出现在启动时?在我的情况下,AF_UNIX套接字文件仍然存在,这破坏了整个想法。在这种情况下,上述通过获取特定文件名锁定的解决方案更加可靠。 - Danylo Gurianov
2
如上所述,该解决方案适用于Linux,但不适用于Mac OS X。 - Bilal and Olga
2
这个解决方案不起作用。我在Ubuntu 14.04上尝试过了。从2个终端窗口同时运行相同的脚本,它们都能正常运行。 - Dimon
2
这不是关于睡眠的问题。代码可以工作,但只能作为内联代码。我正在将其放入函数中。套接字在函数退出时立即消失。 - Steve Cohen
2
'postconnect_gateway_notify_lock' 只是一个任意的字符串,你可以使用 'baa_baa_black_sheep',它仍然可以正常工作。我以为它是一些特定的 Linux 或套接字相关的常量。 - cardamom
显示剩余4条评论

27

我不知道这是否足够符合Python的语言风格,但在Java世界中,监听一个已定义的端口是一种广泛使用的解决方案,因为它适用于所有主要平台,并且不会出现崩溃程序的问题。

监听端口的另一个优点是,您可以向正在运行的实例发送命令。例如,当用户第二次启动程序时,您可以向正在运行的实例发送一个命令,告诉它打开另一个窗口(例如Firefox就是这样做的。虽然我不知道它们是否使用TCP端口或命名管道之类的东西)。


+1,特别是因为它允许我通知正在运行的实例,这样它就可以创建另一个窗口,弹出等。 - WhyNotHugo
2
使用例如 import socket; s = socket.socket(socket.AF_INET, socket.SOCK_STREAM); s.bind(('localhost', DEFINED_PORT))。如果另一个进程绑定到相同的端口,将引发 OSError - crishoj

16

我以前没写过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文件),并在脚本结束(正常或错误)之前保持锁定状态。


1
非常优雅。我对它进行了更改,以便从脚本的参数中获取路径。还建议将其嵌入到某个常见位置 - 示例 - Jossef Harush Kadouri
我发现这个链接很有帮助:link 如果你正在使用Windows,那么fctnl的替代品是win32api。希望这可以帮到你。 - BV_Data

11

使用 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并继续执行。


6
这个程序存在竞态条件。测试-写入序列可能会在两个程序几乎同时开始时引发异常,发现没有文件并尝试并发打开进行写操作。它“应该”在一个程序上引发异常,从而允许另一个程序继续进行。 - S.Lott

8
在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")

有些答案使用fctnl(也包含在@sorin tendo包中),但它在Windows上不可用。如果您尝试使用像pyinstaller这样的软件包冻结Python应用程序并进行静态导入,则会引发错误。

此外,使用锁定文件方法会导致数据库文件出现只读问题(我在使用sqlite3时遇到过)。


1
它似乎对我不起作用(我在Windows 10上运行Python 3.6)。 - Vivian De Smedt

6

以下是我最终的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)

解释

代码尝试创建一个互斥锁,其名称从脚本的完整路径派生而来。我们使用正斜杠来避免与实际文件系统产生混淆。

优点

  • 不需要配置或'魔法'标识符,在尽可能多的不同脚本中使用它。
  • 没有过时的文件留在周围,互斥锁随着您的退出而消失。
  • 等待时打印有用的消息。

5

对于任何使用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()

编写了一个用于单例的套接字列表线程后,我发现这个代码很好用,已经在几个程序中安装了。但是,我希望能够额外添加“唤醒”功能,以便将其置于大量重叠窗口的前台和中心位置。此外,“documented here”链接指向的是无用的自动生成文档,这是更好的链接 - RufusVS
@RufusVS 你说得对 - 那是一个更好的文档链接,我已经更新了答案。 - Matt Coubrough

4

这可能有效。

  1. 尝试在已知位置创建PID文件。如果失败,则有人锁定了该文件,您就完成了。

  2. 正常完成后,关闭并删除PID文件,以便其他人可以覆盖它。

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

您还可以使用PID文件来杀死挂起的程序。


但是如果你的程序中止,它会留下一个PID文件。然后下一次运行脚本时,必须确定它是否是一个过时的PID文件,这可能会导致很多潜在的竞争条件。这就是为什么要使用系统锁定函数的原因。 - undefined

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接