检查python脚本中正在运行的python脚本

您可以尝试使用带有r标志的CalledProcessError -create命令重试指定的次数来捕获CalledProcessError并退出， -p标志将存储进程的pid ：

 import os import sys from time import sleep from subprocess import check_call, CalledProcessError try: check_call(["lockfile-create", "-q","-p", "-r", "0", "-l", "my.lock"]) except CalledProcessError as e: print("{} is already running".format(sys.argv[0])) print(e.returncode) exit(1) # main body for i in range(10): sleep(2) print(1) check_call(["rm","-f","my.lock"]) 

在运行已经运行的代码时，用上面的代码运行test.py脚本会输出以下内容：

 $ python lock.py lock.py is already running 4 

选项

-q， – 安静

抑制任何输出。 只有退出状态才能显示成功或失败。

-v，–verbose

启用诊断输出。

-l，–lock-name

不要将.lock添加到文件名。 此选项适用于lockfile-create，lockfile-remove，lockfile-touch或lockfile-check。

-p，–use-pid

在创建锁定文件时，将当前进程ID（PID）写入锁定文件，并在检查锁定有效性时使用该PID。 有关更多信息，请参见lockfile_create（3）联机帮助页。 此选项适用于lockfile-create，lockfile-remove，lockfile-touch和lockfile-check。

-o，–onehot

触摸锁定并立即退出。 该选项适用于lockfile-touch和mail-touchlock。 如果不提供，这些命令将永远运行，每分钟触摸一次锁，直到死亡。

-r重试次数，–retry重试次数

尝试在放弃之前锁定文件名的重试次数。 每次尝试都会比最后一次（以5秒为增量）延迟一段时间，直到重试之间达到一分钟的最大延迟时间。 如果未指定重试计数，则默认值为9，如果所有9次锁定尝试都失败，则会在180秒（3分钟）后放弃。

描述

lockfile_create函数以NFS安全方式创建一个锁文件。

如果标志设置为L_PID，那么lockfile_create将不仅检查现有的锁文件，而且还将读取内容以查看它是否包含ASCII中的进程标识。 如果是这样，锁定文件只有在该进程仍然存在的情况下才有效。

如果锁定文件位于共享文件系统上，则可能是由远程主机上的进程创建的。 因此，进程ID检查是无用的，不应该设置L_PID标志。 在这种情况下，没有什么好方法来查看锁文件是否陈旧。 因此，如果锁定文件大于5分钟，它将被删除。 这就是为什么提供lockfile_touch函数的原因：在持有锁的同时，需要通过调用lockfile_touch（）定期刷新（每分钟左右）。

lockfile_check函数检查一个有效的锁文件是否已经存在，而不尝试创建一个新的锁文件。

最后，lockfile_remove函数删除锁定文件。

算法

用于以原子方式（甚至通过NFS）创建锁定文件的算法如下所示：

1

一个独特的文件被创建。 在printf格式中，文件的名称是.lk％05d％x％s。 第一个参数（％05d）是当前的进程ID。 第二个参数（％x）由时间（2）返回的值的4个小比特组成。 最后一个参数是系统主机名。

2

然后使用链接（2）创建锁定文件。 链接的返回值被忽略。

3

现在lockfile是stat（）。 如果统计失败，我们进入第6步。

4

将lockfile的stat值与临时文件的stat值进行比较。 如果他们是一样的，我们有锁。 临时文件被删除，值0（成功）返回给调用者。

五

检查现有的锁文件是否是有效的。 如果无效，则删除旧的锁定文件。

6

在重试之前，我们睡了n秒。 n最初是5秒，但是在每次重试之后，5个额外的秒被添加到最多60秒（增量退避）。 然后我们继续步骤2重试次数。

在redhat上似乎有一个名为lockfile-progs的等价包。

在Mac上，您可以使用lockfile并执行如下操作：

 import os import sys from time import sleep import os from subprocess import Popen, CalledProcessError, check_call p = Popen(["lockfile", "-r", "0", "my.lock"]) p.wait() if p.returncode == 0: with open("my.pid", "w") as f: f.write(str(os.getpid())) else: try: with open("my.pid") as f: # see if process is still running or lockfile # is left over from previous run. r = f.read() check_call(["kill", "-0", "{}".format(r)]) except CalledProcessError: # remove old lock file and create new check_call(["rm", "-f", "my.lock"]) check_call(["lockfile", "-r", "0", "my.lock"]) # update pid with open("my.pid", "w") as out: out.write(str(os.getpid())) print("Deleted stale lockfile.") else: print("{} is already running".format(sys.argv[0])) print(p.returncode) exit(1) # main body for i in range(10): sleep(1) print(1) check_call(["rm", "-f", "my.lock"]) 

在你的情况下，也许使用套接字将工作：

 from socket import socket, gethostname, error, SO_REUSEADDR, SOL_SOCKET from sys import argv import errno sock = socket() # Create a socket object host = gethostname() # /proc/sys/net/ipv4/ip_local_port_range is 32768 61000 on my Ubuntu Machine port = 60001 # allow connection in TIME_WAIT sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) try: sock.bind((host, port)) sock.connect((host, port)) except error as e: # [Errno 99] Cannot assign requested address if e.errno == errno.EADDRNOTAVAIL: print("{} is already running".format(argv[0])) exit(1) # else raise the error else: raise e # main body from time import sleep while True: print(1) sleep(2) sock.close()