Python案例如何实现进程守护？

wen python案例 2026-06-07 76

本文目录导读：

Python案例如何实现进程守护？

使用标准库 Daemon（推荐）——统一身份认证服务守护
手动实现守护进程（不依赖第三方库）
使用 supervisor 管理进程（最实用）
使用 subprocess + 信号处理（适合简单监控）
最佳实践建议

在Python中实现进程守护（守护进程）有多种方法，根据需求不同可以选择不同的解决方案，以下是几种常见且实用的实现方式：

使用标准库 `Daemon`（推荐）——统一身份认证服务守护

python-daemon 是最常用的进程守护库。

安装

pip install python-daemon

示例：文件监控守护进程

import daemon
import time
import os
import logging
import sys
# 日志配置
logging.basicConfig(
    filename='/var/log/mydaemon.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
# 自定义工作函数
def run_worker():
    while True:
        # 模拟工作——检查某个文件的变化
        current_time = time.strftime("%Y-%m-%d %H:%M:%S")
        logging.info(f"守护进程正在运行，时间: {current_time}")
        # 这里放实际的工作逻辑
        # 比如检查文件、监控进程、处理数据等
        time.sleep(10)
# 守护进程上下文
def start_daemon():
    # 设置工作目录、PID文件、日志等
    with daemon.DaemonContext(
        working_directory='/opt/mydaemon',
        umask=0o002,
        pidfile=daemon.pidfile.TimeoutPIDLockFile('/var/run/mydaemon.pid'),
        stdout=sys.stdout,
        stderr=sys.stderr
    ):
        run_worker()
if __name__ == '__main__':
    start_daemon()

手动实现守护进程（不依赖第三方库）

import os
import sys
import time
import logging
def daemonize(stdin='/dev/null', stdout='/dev/null', stderr='/dev/null'):
    """完全分离进程并转为守护进程"""
    # 第一步：fork并退出父进程
    try:
        pid = os.fork()
        if pid > 0:
            sys.exit(0)  # 父进程退出
    except OSError as e:
        sys.stderr.write(f"fork #1 失败: {e}\n")
        sys.exit(1)
    # 分离子进程与父进程的控制终端
    os.setsid()
    os.umask(0o22)
    # 第二步：再次fork确保进程不会重新获得终端
    try:
        pid = os.fork()
        if pid > 0:
            sys.exit(0)
    except OSError as e:
        sys.stderr.write(f"fork #2 失败: {e}\n")
        sys.exit(1)
    # 重定向标准文件描述符
    stdin = open(stdin, 'r')
    stdout = open(stdout, 'a+')
    stderr = open(stderr, 'a+')
    os.dup2(stdin.fileno(), 0)
    os.dup2(stdout.fileno(), 1)
    os.dup2(stderr.fileno(), 2)
    # 关闭文件描述符
    for fd in range(3, 256):
        try:
            os.close(fd)
        except OSError:
            pass
# 实际的守护进程工作类
class MyDaemon:
    def __init__(self, pidfile):
        self.pidfile = pidfile
    def run(self):
        while True:
            # 工作逻辑
            with open('/tmp/daemon_status.log', 'a') as f:
                f.write(f"守护进程运行中: {time.ctime()}\n")
            time.sleep(10)
    def start(self):
        # 检查是否已有PID文件
        if os.path.exists(self.pidfile):
            with open(self.pidfile, 'r') as f:
                pid = int(f.read())
            if os.path.exists(f'/proc/{pid}'):
                print("进程已在运行中")
                return
        daemonize('/dev/null', '/tmp/daemon_stdout.log', '/tmp/daemon_stderr.log')
        # 写PID文件
        with open(self.pidfile, 'w') as f:
            f.write(str(os.getpid()))
        self.run()
    def stop(self):
        if os.path.exists(self.pidfile):
            with open(self.pidfile, 'r') as f:
                pid = int(f.read())
            os.kill(pid, 15)  # 发送SIGTERM
            os.remove(self.pidfile)
            print(f"已停止进程 {pid}")
        else:
            print("进程未运行")
# 使用示例
if __name__ == '__main__':
    daemon = MyDaemon('/tmp/mydaemon.pid')
    if len(sys.argv) != 2:
        print("用法: python daemon.py [start|stop|restart]")
        sys.exit(1)
    action = sys.argv[1]
    if action == 'start':
        daemon.start()
    elif action == 'stop':
        daemon.stop()
    elif action == 'restart':
        daemon.stop()
        time.sleep(1)
        daemon.start()
    else:
        print(f"未知命令: {action}")

使用 `supervisor` 管理进程（最实用）

虽然supervisor不是纯Python实现,但它是最常用的进程管理系统：

配置 supervisord

# /etc/supervisor/conf.d/myapp.conf
[program:myapp]
command=python /opt/myapp/main.py
directory=/opt/myapp
user=www-data
autostart=true
autorestart=true
startretries=3
stderr_logfile=/var/log/myapp.err.log
stdout_logfile=/var/log/myapp.out.log

Python代码（无需特殊处理）

# main.py
import time
import logging
logging.basicConfig(level=logging.INFO)
def main():
    while True:
        logging.info("应用正在运行...")
        time.sleep(10)
if __name__ == '__main__':
    main()

使用 `subprocess` + 信号处理（适合简单监控）

import subprocess
import signal
import os
import sys
import time
class ProcessGuardian:
    def __init__(self, command):
        self.command = command
        self.process = None
        self.running = True
    def start_process(self):
        print(f"启动进程: {self.command}")
        self.process = subprocess.Popen(
            self.command,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )
    def monitor(self):
        # 设置信号处理
        signal.signal(signal.SIGINT, self.handle_exit)
        signal.signal(signal.SIGTERM, self.handle_exit)
        while self.running:
            # 检查进程状态
            if self.process.poll() is not None:
                print(f"进程退出，代码: {self.process.returncode}")
                self.start_process()  # 自动重启
            time.sleep(5)
    def handle_exit(self, signum, frame):
        print("收到退出信号...")
        self.running = False
        if self.process:
            self.process.terminate()
        sys.exit(0)
if __name__ == '__main__':
    guardian = ProcessGuardian("python worker.py")
    guardian.start_process()
    guardian.monitor()