我有一个作为我的 Web 应用程序一部分运行的 Python 守护进程。如何快速检查是否正在运行,如果没有,则启动它?
我想这样做是为了解决守护进程的任何崩溃,并且脚本不必手动运行,只要调用即可自动运行并保持运行。
我该如何使用 Python 检查我的脚本是否正在运行?
我有一个作为我的 Web 应用程序一部分运行的 Python 守护进程。如何快速检查是否正在运行,如果没有,则启动它?
我想这样做是为了解决守护进程的任何崩溃,并且脚本不必手动运行,只要调用即可自动运行并保持运行。
我该如何使用 Python 检查我的脚本是否正在运行?
在Linux系统上很方便的一种技术是使用域套接字:
import socket
import sys
import time
def get_lock(process_name):
# Without holding a reference to our socket somewhere it gets garbage
# collected when the function exits
get_lock._lock_socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
try:
# The null byte (\0) means the socket is created
# in the abstract namespace instead of being created
# on the file system itself.
# Works only in Linux
get_lock._lock_socket.bind('\0' + process_name)
print 'I got the lock'
except socket.error:
print 'lock exists'
sys.exit()
get_lock('running_test')
while True:
time.sleep(3)
使用原子操作可以避免如果进程被发送 SIGKILL
信号时,锁文件散落在各处的问题。
您可以在 socket.close的文档 中了解到,当垃圾回收时,套接字会自动关闭。
\0
)表示套接字在抽象命名空间中创建,而不是在文件系统本身上创建。 - aychedee在某个地方放置一个pid文件(例如/tmp)。然后,您可以通过检查文件中的PID是否存在来检查进程是否正在运行。在干净关闭时不要忘记删除该文件,并在启动时检查它。
#/usr/bin/env python
import os
import sys
pid = str(os.getpid())
pidfile = "/tmp/mydaemon.pid"
if os.path.isfile(pidfile):
print "%s already exists, exiting" % pidfile
sys.exit()
file(pidfile, 'w').write(pid)
try:
# Do some actual work here
finally:
os.unlink(pidfile)
你可以检查/tmp/mydaemon.pid文件中的内容是否为一个存在的进程,以确定该进程是否正在运行。可以使用上面提到的Monit来完成此操作,也可以编写一个简单的shell脚本并使用ps命令的返回代码来检查它。
ps up `cat /tmp/mydaemon.pid ` >/dev/null && echo "Running" || echo "Not running"
如果您想额外加分,可以使用atexit模块确保您的程序在任何情况下(比如被杀死、引发异常等)都能清理其pid文件。
if os.path.isfile(pidfile)
可能会对两者都评估为false,导致它们都写入锁定文件并继续运行。 - Cerinfile()
已经被移除了,需要使用open()
。即使您使用的是Python 2.7版本,也应该使用open()
而非file()
,详情请见:https://docs.python.org/2/library/functions.html#file (是的,如果您使用的是早期的Python 2.2版本,则官方建议与现在相反。显然,他们改变了主意。) - jpkpid库可以准确地完成这个任务。
from pid import PidFile
with PidFile():
do_something()
它还将自动处理pid文件存在但进程未运行的情况。PidFile(piddir='/home/user/run/')
等方式来指定一个你有权限放置pid文件的目录。这样你就不需要以root身份运行了。 - Decko我的解决方案是检查进程和命令行参数 在Windows和Ubuntu Linux上测试过
import psutil
import os
def is_running(script):
for q in psutil.process_iter():
if q.name().startswith('python'):
if len(q.cmdline())>1 and script in q.cmdline()[1] and q.pid !=os.getpid():
print("'{}' Process is already running".format(script))
return True
return False
if not is_running("test.py"):
n = input("What is Your Name? ")
print ("Hello " + n)
python ..
启动的,而不是直接调用 ./<script name>
,否则它将无法工作,因为它检查进程是否以 python
开始。 - DMin当然,来自Dan的示例不会按照应有的方式运行。
事实上,如果脚本崩溃、引发异常或无法清理pid文件,则脚本将运行多次。
我建议基于另一个网站的以下内容:
这是为了检查是否已经存在锁定文件。
\#/usr/bin/env python
import os
import sys
if os.access(os.path.expanduser("~/.lockfile.vestibular.lock"), os.F_OK):
#if the lockfile is already there then check the PID number
#in the lock file
pidfile = open(os.path.expanduser("~/.lockfile.vestibular.lock"), "r")
pidfile.seek(0)
old_pid = pidfile.readline()
# Now we check the PID from lock file matches to the current
# process PID
if os.path.exists("/proc/%s" % old_pid):
print "You already have an instance of the program running"
print "It is running as process %s," % old_pid
sys.exit(1)
else:
print "File is there but the program is not running"
print "Removing lock file for the: %s as it can be there because of the program last time it was run" % old_pid
os.remove(os.path.expanduser("~/.lockfile.vestibular.lock"))
这是放置PID文件在锁文件中的代码部分
pidfile = open(os.path.expanduser("~/.lockfile.vestibular.lock"), "w")
pidfile.write("%s" % os.getpid())
pidfile.close()
这段代码将检查pid的值是否与已有运行中的进程相比,以避免重复执行。
希望这能有所帮助。
os.kill(old_pid, 0)
,它在各种UNIX系统中更加通用。如果PID不存在或属于不同的用户,它会引发OSError
异常。 - drdaeman在寻找解决方法时,我偶然发现了这个老问题。
使用psutil:
import psutil
import sys
from subprocess import Popen
for process in psutil.process_iter():
if process.cmdline() == ['python', 'your_script.py']:
sys.exit('Process found: exiting.')
print('Process not found: starting it.')
Popen(['python', 'your_script.py'])
在UNIX系统中,有非常好的包可以用来重新启动进程。其中一个拥有出色教程的软件是monit。通过一些调整,你可以使用这个经过验证的可靠技术来保持你的守护进程稳定运行。
有很多选择。一种方法是使用系统调用或Python库来替您执行此类调用。另一种方法则是简单地生成一个进程,例如:
ps ax | grep processName
并解析输出。许多人选择这种方法,在我看来并不一定是一个坏方法。
multiprocessing.shared_memory
的可移植解决方案:import atexit
from multiprocessing import shared_memory
_ensure_single_process_store = {}
def ensure_single_process(name: str):
if name in _ensure_single_process_store:
return
try:
shm = shared_memory.SharedMemory(name='ensure_single_process__' + name,
create=True,
size=1)
except FileExistsError:
print(f"{name} is already running!")
raise
_ensure_single_process_store[name] = shm
atexit.register(shm.unlink)
atexit
,但有时在异常退出时清理会很有帮助。试试这个其他版本
def checkPidRunning(pid):
'''Check For the existence of a unix pid.
'''
try:
os.kill(pid, 0)
except OSError:
return False
else:
return True
# Entry point
if __name__ == '__main__':
pid = str(os.getpid())
pidfile = os.path.join("/", "tmp", __program__+".pid")
if os.path.isfile(pidfile) and checkPidRunning(int(file(pidfile,'r').readlines()[0])):
print "%s already exists, exiting" % pidfile
sys.exit()
else:
file(pidfile, 'w').write(pid)
# Do some actual work here
main()
os.unlink(pidfile)