确保只运行一个程序实例

问题:确保只运行一个程序实例

有没有一种Python方式可以只运行一个程序实例?

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

(考虑到程序有时可能会失败,例如segfault-因此“锁定文件”之类的东西将无法工作)

Is there a Pythonic way to have only one instance of a program running?

The only reasonable solution I’ve come up with is trying to run it as a server on some port, then second program trying to bind to same port – fails. But it’s not really a great idea, maybe there’s something more lightweight than this?

(Take into consideration that program is expected to fail sometimes, i.e. segfault – so things like “lock file” won’t work)


回答 0

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

The following code should do the job, it is cross-platform and runs on Python 2.4-3.2. I tested it on Windows, OS X and Linux.

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

The latest code version is available singleton.py. Please file bugs here.

You can install tend using one of the following methods:


回答 1

简单,跨平台的解决方案,在发现了另一个问题Zgoda酒店

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

Simple, cross-platform solution, found in another question by zgoda:

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

A lot like S.Lott’s suggestion, but with the code.


回答 2

此代码特定于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可以更改唯一字符串,以允许需要强制执行单个实例的多个程序。

This code is Linux specific. It uses ‘abstract’ UNIX domain sockets, but it is simple and won’t leave stale lock files around. I prefer it to the solution above because it doesn’t require a specially reserved TCP port.

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) 

The unique string postconnect_gateway_notify_lock can be changed to allow multiple programs that need a single instance enforced.


回答 3

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

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

I don’t know if it’s pythonic enough, but in the Java world listening on a defined port is a pretty widely used solution, as it works on all major platforms and doesn’t have any problems with crashing programs.

Another advantage of listening to a port is that you could send a command to the running instance. For example when the users starts the program a second time, you could send the running instance a command to tell it to open another window (that’s what Firefox does, for example. I don’t know if they use TCP ports or named pipes or something like that, ‘though).


回答 4

以前从未编写过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文件)并保持锁定状态,直到脚本结束(正常或错误)。

Never written python before, but this is what I’ve just implemented in mycheckpoint, to prevent it being started twice or more by 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()

Found Slava-N’s suggestion after posting this in another issue (http://stackoverflow.com/questions/2959474). This one is called as a function, locks the executing scripts file (not a pid file) and maintains the lock until the script ends (normal or error).


回答 5

使用一个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())

因此,换句话说,您正在检查是否存在pidfile。如果不是,请将您的pid写入该文件。如果pidfile存在,则检查pid是否为正在运行的进程的pid;如果是这样,则您有另一个正在运行的实时进程,因此只需关闭。如果不是,则先前的进程崩溃了,因此将其记录下来,然后将您自己的pid写入旧文件中。然后继续。

Use a pid file. You have some known location, “/path/to/pidfile” and at startup you do something like this (partially pseudocode because I’m pre-coffee and don’t want to work all that hard):

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())

So, in other words, you’re checking if a pidfile exists; if not, write your pid to that file. If the pidfile does exist, then check to see if the pid is the pid of a running process; if so, then you’ve got another live process running, so just shut down. If not, then the previous process crashed, so log it, and then write your own pid to the file in place of the old one. Then continue.


回答 6

您已经在另一个线程中找到了对类似问题的答复,因此为了完整起见,请参见如何在Windows上实现名为Mutex的相同目的。

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

You already found reply to similar question in another thread, so for completeness sake see how to achieve the same on Windows uning named mutex.

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


回答 7

这可能有效。

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

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

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

如果程序挂起,也可以使用PID文件将其杀死。

This may work.

  1. Attempt create a PID file to a known location. If you fail, someone has the file locked, you’re done.

  2. When you finish normally, close and remove the PID file, so someone else can overwrite it.

You can wrap your program in a shell script that removes the PID file even if your program crashes.

You can, also, use the PID file to kill the program if it hangs.


回答 8

在UNIX上,使用锁定文件是一种非常普遍的方法。如果崩溃,则必须手动清理。您可以将PID存储在文件中,并在启动时检查是否有使用此PID的进程,否则将覆盖锁定文件。(但是,您还需要在read-file-check-pid-rewrite-file周围加锁)。您将在os -package中找到获取和检查pid所需的内容。检查是否存在具有给定pid的进程的常见方法是向其发送非致命信号。

其他替代方法可以将其与羊群或posix信号量结合使用。

如saua所建议的那样,打开网络插座可能是最简单,最方便的方法。

Using a lock-file is a quite common approach on unix. If it crashes, you have to clean up manually. You could stor the PID in the file, and on startup check if there is a process with this PID, overriding the lock-file if not. (However, you also need a lock around the read-file-check-pid-rewrite-file). You will find what you need for getting and checking pid in the os-package. The common way of checking if there exists a process with a given pid, is to send it a non-fatal signal.

Other alternatives could be combining this with flock or posix semaphores.

Opening a network socket, as saua proposed, would probably be the easiest and most portable.


回答 9

对于将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

这是一个简单的直接替换wx.App,禁止多个实例。要使用它,只需在代码中将替换wx.AppSingleApp,如下所示:

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

For anybody using wxPython for their application, you can use the function wx.SingleInstanceChecker documented here.

I personally use a subclass of wx.App which makes use of wx.SingleInstanceChecker and returns False from OnInit() if there is an existing instance of the app already executing like so:

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

This is a simple drop-in replacement for wx.App that prohibits multiple instances. To use it simply replace wx.App with SingleApp in your code like so:

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

回答 10

这是我最终只能使用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)

说明

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

优点

  • 不需要配置或“魔术”标识符,请根据需要在许多不同的脚本中使用它。
  • 周围没有陈旧的文件,互斥锁将与您一起消亡。
  • 等待时打印有用的消息

Here is my eventual Windows-only solution. Put the following into a module, perhaps called ‘onlyone.py’, or whatever. Include that module directly into your __ main __ python script file.

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)

Explanation

The code attempts to create a mutex with name derived from the full path to the script. We use forward-slashes to avoid potential confusion with the real file system.

Advantages

  • No configuration or ‘magic’ identifiers needed, use it in as many different scripts as needed.
  • No stale files left around, the mutex dies with you.
  • Prints a helpful message when waiting

回答 11

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了Windows上不可用的(也包含在@sorin tento软件包中),并且如果您尝试使用像pyinstaller静态导入这样的软件包来冻结python应用程序,则会引发错误。

此外,使用锁定文件方法还会导致read-only数据库文件出现问题(sqlite3)。

The best solution for this on windows is to use mutexes as suggested by @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")

Some answers use fctnl (included also in @sorin tendo package) which is not available on windows and should you try to freeze your python app using a package like pyinstaller which does static imports, it throws an error.

Also, using the lock file method, creates a read-only problem with database files( experienced this with sqlite3).


回答 12

我将其发布为答案,因为我是新用户,并且Stack Overflow尚未允许我投票。

Sorin Sbarnea的解决方案可在OS X,Linux和Windows下为我工作,对此我深表感谢。

但是,tempfile.gettempdir()在OS X和Windows下表现为一种方式,而在其他some / many / all(?)* nixes下表现为另一种方式(忽略OS X也是Unix!)。区别对于此代码很重要。

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

这意味着在多用户计算机上创建锁文件时,它是在/ tmp中创建的,只有第一次创建锁文件的用户才能运行该应用程序。

一种可能的解决方案是将当前用户名嵌入到锁定文件的名称中。

值得注意的是,OP抢占端口的解决方案在多用户计算机上也会出现异常。

I’m posting this as an answer because I’m a new user and Stack Overflow won’t let me vote yet.

Sorin Sbarnea’s solution works for me under OS X, Linux and Windows, and I am grateful for it.

However, tempfile.gettempdir() behaves one way under OS X and Windows and another under other some/many/all(?) *nixes (ignoring the fact that OS X is also Unix!). The difference is important to this code.

OS X and Windows have user-specific temp directories, so a tempfile created by one user isn’t visible to another user. By contrast, under many versions of *nix (I tested Ubuntu 9, RHEL 5, OpenSolaris 2008 and FreeBSD 8), the temp dir is /tmp for all users.

That means that when the lockfile is created on a multi-user machine, it’s created in /tmp and only the user who creates the lockfile the first time will be able to run the application.

A possible solution is to embed the current username in the name of the lock file.

It’s worth noting that the OP’s solution of grabbing a port will also misbehave on a multi-user machine.


回答 13

single_process在我的gentoo上使用;

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

I use single_process on my gentoo;

pip install single_process

example:

from single_process import single_process

@single_process
def main():
    print 1

if __name__ == "__main__":
    main()   

refer: https://pypi.python.org/pypi/single_process/1.0


回答 14

我一直怀疑使用进程组应该有一个不错的POSIXy解决方案,而不必使用文件系统,但是我不太确定。就像是:

启动时,您的进程会向特定组中的所有进程发送“ kill -0”。如果存在任何此类进程,则退出。然后,它加入该组。没有其他进程使用该组。

但是,这是一个竞争条件-多个进程都可以恰好同时执行此操作,并且最终都加入了该团队并同时运行。到您添加某种互斥锁使其具有水密性时,您不再需要过程组。

如果您的过程仅由cron启动,每分钟或每小时启动一次,这可能是可以接受的,但是这让我有些紧张,因为它恰恰在您不希望的那一天出错。

我猜毕竟这不是一个很好的解决方案,除非有人可以改进?

I keep suspecting there ought to be a good POSIXy solution using process groups, without having to hit the file system, but I can’t quite nail it down. Something like:

On startup, your process sends a ‘kill -0’ to all processes in a particular group. If any such processes exist, it exits. Then it joins the group. No other processes use that group.

However, this has a race condition – multiple processes could all do this at precisely the same time and all end up joining the group and running simultaneously. By the time you’ve added some sort of mutex to make it watertight, you no longer need the process groups.

This might be acceptable if your process only gets started by cron, once every minute or every hour, but it makes me a bit nervous that it would go wrong precisely on the day when you don’t want it to.

I guess this isn’t a very good solution after all, unless someone can improve on it?


回答 15

上周,我遇到了这个确切的问题,尽管我确实找到了一些好的解决方案,但我还是决定制作一个非常简单干净的python软件包并将其上传到PyPI。它与tendo的不同之处在于它可以锁定任何字符串资源名称。尽管您当然可以锁定__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

I ran into this exact problem last week, and although I did find some good solutions, I decided to make a very simple and clean python package and uploaded it to PyPI. It differs from tendo in that it can lock any string resource name. Although you could certainly lock __file__ to achieve the same effect.

Install with: pip install quicklock

Using it is extremely simple:

[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'!

Take a look: https://pypi.python.org/pypi/quicklock


回答 16

在罗伯托·罗萨里奥(Roberto Rosario)的回答的基础上,我提出了以下功能:

SOCKET = None
def run_single_instance(uniq_name):
    try:
        import socket
        global SOCKET
        SOCKET = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        ## Create an abstract socket, by prefixing it with null.
        # this relies on a feature only in linux, when current process quits, the
        # socket will be deleted.
        SOCKET.bind('\0' + uniq_name)
        return True
    except socket.error as e:
        return False

我们需要定义全局SOCKET变量,因为只有在整个过程退出时才会对其进行垃圾收集。如果我们在函数中声明局部变量,则在函数退出后它将超出范围,因此将删除套接字。

所有的功劳应该归功于罗伯托·罗萨里奥(Roberto Rosario),因为我只是澄清和阐述了他的代码。而且此代码仅在Linux上有效,如https://troydhanson.github.io/network/Unix_domain_sockets.html中以下引用的文字所述:

Linux具有一项特殊功能:如果UNIX域套接字的路径名以空字节\ 0开头,则其名称不会映射到文件系统中。因此,它不会与文件系统中的其他名称冲突。同样,当服务器在抽象命名空间中关闭其UNIX域侦听套接字时,其文件也将被删除。使用常规的UNIX域套接字,该文件在服务器关闭后仍然存在。

Building upon Roberto Rosario’s answer, I come up with the following function:

SOCKET = None
def run_single_instance(uniq_name):
    try:
        import socket
        global SOCKET
        SOCKET = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        ## Create an abstract socket, by prefixing it with null.
        # this relies on a feature only in linux, when current process quits, the
        # socket will be deleted.
        SOCKET.bind('\0' + uniq_name)
        return True
    except socket.error as e:
        return False

We need to define global SOCKET vaiable since it will only be garbage collected when the whole process quits. If we declare a local variable in the function, it will go out of scope after the function exits, thus the socket be deleted.

All the credit should go to Roberto Rosario, since I only clarify and elaborate upon his code. And this code will work only on Linux, as the following quoted text from https://troydhanson.github.io/network/Unix_domain_sockets.html explains:

Linux has a special feature: if the pathname for a UNIX domain socket begins with a null byte \0, its name is not mapped into the filesystem. Thus it won’t collide with other names in the filesystem. Also, when a server closes its UNIX domain listening socket in the abstract namespace, its file is deleted; with regular UNIX domain sockets, the file persists after the server closes it.


回答 17

linux示例

此方法是基于创建临时文件后关闭应用程序而自动删除的。程序启动后,我们验证文件是否存在;如果文件存在(正在执行中),则程序关闭;否则,它将创建文件并继续执行程序。

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

linux example

This method is based on the creation of a temporary file automatically deleted after you close the application. the program launch we verify the existence of the file; if the file exists ( there is a pending execution) , the program is closed ; otherwise it creates the file and continues the execution of the program.

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

回答 18

在Linux系统上,还可以询问 pgrep -a实例数,该脚本位于进程列表中(选项-a显示完整的命令行字符串)。例如

import os
import sys
import subprocess

procOut = subprocess.check_output( "/bin/pgrep -u $UID -a python", shell=True, 
                                   executable="/bin/bash", universal_newlines=True)

if procOut.count( os.path.basename(__file__)) > 1 :        
    sys.exit( ("found another instance of >{}<, quitting."
              ).format( os.path.basename(__file__)))

-u $UID如果限制适用于所有用户,请删除。免责声明:a)假定脚本的(基本)名称是唯一的,b)可能存在竞争条件。

On a Linux system one could also ask pgrep -a for the number of instances, the script is found in the process list (option -a reveals the full command line string). E.g.

import os
import sys
import subprocess

procOut = subprocess.check_output( "/bin/pgrep -u $UID -a python", shell=True, 
                                   executable="/bin/bash", universal_newlines=True)

if procOut.count( os.path.basename(__file__)) > 1 :        
    sys.exit( ("found another instance of >{}<, quitting."
              ).format( os.path.basename(__file__)))

Remove -u $UID if the restriction should apply to all users. Disclaimer: a) it is assumed that the script’s (base)name is unique, b) there might be race conditions.


回答 19

import sys,os

# start program
try:  # (1)
    os.unlink('lock')  # (2)
    fd=os.open("lock", os.O_CREAT|os.O_EXCL) # (3)  
except: 
    try: fd=os.open("lock", os.O_CREAT|os.O_EXCL) # (4) 
    except:  
        print "Another Program running !.."  # (5)
        sys.exit()  

# your program  ...
# ...

# exit program
try: os.close(fd)  # (6)
except: pass
try: os.unlink('lock')  
except: pass
sys.exit()  
import sys,os

# start program
try:  # (1)
    os.unlink('lock')  # (2)
    fd=os.open("lock", os.O_CREAT|os.O_EXCL) # (3)  
except: 
    try: fd=os.open("lock", os.O_CREAT|os.O_EXCL) # (4) 
    except:  
        print "Another Program running !.."  # (5)
        sys.exit()  

# your program  ...
# ...

# exit program
try: os.close(fd)  # (6)
except: pass
try: os.unlink('lock')  
except: pass
sys.exit()