做开发、运维、数据分析的同学，大概率都有过这些刻骨铭心的经历：
- 半夜手机突然响起，线上服务因为配置文件被误改、进程异常退出，导致业务卡死，不得不爬起来远程改配置、重启服务；
- 定时运行的爬虫/数据同步脚本，因为网络波动、端口占用、内存溢出，悄无声息地挂掉，第二天发现数据漏采、任务中断，只能补跑返工；
- 服务器磁盘满了、日志文件撑爆分区、数据库连接数耗尽，程序报错但无人知晓，直到用户反馈才后知后觉；
- 测试/生产环境的核心程序，需要7*24小时运行，却总要人工巡检，稍不注意就出故障，身心俱疲。
这些问题的核心痛点只有一个：所有的服务/程序/配置的异常，都需要「人工介入」处理，而且往往是故障发生后被动补救。
我们追求的理想状态是：程序能自己发现问题、自己解决问题，不用人盯着，半夜再也不用被叫醒。
今天，我就带大家从零到一，用 Python 写一个万能的「看门狗」（WatchDog）脚本。这个脚本轻量无依赖、无需部署复杂框架、可直接运行、按需修改，还会针对工作中6个高频故障场景给出「专属解决方案+落地代码」，覆盖90%的日常运维需求。从此告别半夜爬起来改配置、重启服务的日子，彻底解放双手！
核心价值：自动化监控 + 自动化检测 + 自动化修复，一套脚本，搞定所有「需要人工兜底」的服务异常问题。
一、什么是「看门狗」脚本？它的核心能力是什么？
看门狗(WatchDog)的定义
看门狗脚本，本质是 一个后台持续运行的「守护检测程序」，它就像一个忠诚的保安，7*24小时盯着你的业务程序、服务器配置、核心服务，循环执行「检测 → 发现异常 → 自动修复 → 告警通知」的闭环流程。
它不是什么高大上的技术，却是性价比最高的运维利器，完全用 Python 原生语法实现，不需要安装任何重量级框架，几百行代码就能解决大问题。
看门狗脚本的核心能力（四大金刚）
- 持续监控：后台常驻运行，无限循环执行检测逻辑，间隔可自定义（比如每10秒检测一次）；
- 异常检测：精准识别「进程挂了、配置错了、端口占用、磁盘满了、日志报错、网络不通」等各类异常；
- 自动修复：发现异常后，无需人工干预，自动执行「重启进程、恢复配置、释放端口、清理日志」等修复动作；
- 告警通知：修复完成/修复失败/严重异常时，自动推送告警（邮件、企业微信/钉钉、短信），做到「有事知会，无事不扰」。
为什么用 Python 写看门狗？
- 跨平台：一套代码，在 Windows、Linux、Mac 上都能运行，不管是本地服务还是云服务器，无缝适配；
- 语法简洁：Python 对文件操作、进程管理、系统命令、网络请求的封装都极其友好，几行代码就能实现核心逻辑；
- 无依赖：核心功能全部用 Python 标准库实现，不用安装第三方包，部署到任何机器都能直接跑；
- 灵活易改：脚本轻量化，根据业务需求改几行配置、加一段检测逻辑，就能适配新的场景，运维成本极低。
二、看门狗脚本的通用核心框架（必学！所有场景都基于此）
所有的看门狗脚本，都遵循一个通用的核心逻辑，这是骨架，所有场景的解决方案都是在这个骨架上「添肉」。先吃透这个框架，后面的场景改造会非常轻松。
✅ 核心设计思路（极简，5步闭环）
初始化配置 → 无限循环检测 → 发现异常 → 执行自动修复 → 记录日志/发送告警
✅ Python 通用看门狗基础模板（可直接复用，无任何依赖）
这个模板是万能的，包含了「日志记录、异常捕获、循环检测、自定义配置」等所有基础能力，后面所有场景的解决方案，都是基于这个模板修改而来。
import os
import time
import logging
import subprocess
from datetime import datetime
# ===================== 1. 基础配置（按需修改，核心参数） =====================
# 检测间隔时间(秒)，比如每10秒检测一次
CHECK_INTERVAL = 10
# 日志文件路径，所有检测/修复日志都会写入，方便排查问题
LOG_FILE = "./watchdog_monitor.log"
# 需要监控的目标程序/服务/配置相关参数，后续场景按需修改
TARGET_CONFIG = {
"process_name": "python", # 目标进程名
"config_path": "./config.ini", # 目标配置文件路径
"service_cmd": "python main.py"# 重启服务的命令
}
# ===================== 2. 日志配置（重要！所有操作留痕，方便排查） =====================
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[
logging.FileHandler(LOG_FILE, encoding="utf-8"), # 写入日志文件
logging.StreamHandler() # 控制台打印
]
)
logger = logging.getLogger(__name__)
# ===================== 3. 看门狗核心基类（通用逻辑，无需修改） =====================
classBaseWatchDog:
def__init__(self):
self.check_interval = CHECK_INTERVAL
self.target_config = TARGET_CONFIG
logger.info("【看门狗启动成功】开始7*24小时监控，检测间隔：{}秒".format(self.check_interval))
defcheck_status(self):
"""【核心抽象方法】检测状态，子类按需实现具体的检测逻辑"""
raise NotImplementedError("子类必须实现检测逻辑")
defrepair(self):
"""【核心抽象方法】异常修复，子类按需实现具体的修复逻辑"""
raise NotImplementedError("子类必须实现修复逻辑")
defsend_alert(self, msg):
"""【告警通知】可扩展：邮件/钉钉/企业微信/短信，基础版打印日志"""
logger.warning("【告警通知】{}".format(msg))
defrun(self):
"""【看门狗主循环】无限运行，核心入口"""
whileTrue:
try:
# 第一步：执行检测逻辑
is_normal = self.check_status()
if is_normal:
logger.info("检测正常，一切安好✔️")
else:
logger.error("检测发现异常❌，开始执行自动修复...")
# 第二步：执行修复逻辑
repair_result = self.repair()
if repair_result:
logger.info("自动修复成功✅")
self.send_alert("异常已自动修复完成，服务恢复正常")
else:
logger.critical("自动修复失败❌，请人工介入处理！")
self.send_alert("【紧急告警】自动修复失败，服务异常，请立即处理！")
except Exception as e:
logger.error("看门狗执行检测时发生错误：{}".format(str(e)), exc_info=True)
# 间隔指定时间，再次检测
time.sleep(self.check_interval)
# ===================== 启动看门狗 =====================
if __name__ == "__main__":
# 后续每个场景，替换这里的子类即可
watchdog = BaseWatchDog()
watchdog.run()
模板核心亮点
- 高内聚低耦合：检测逻辑和修复逻辑完全分离，新增场景只需继承基类，实现两个方法即可；
- 完整日志记录：所有操作都有日志，出问题能快速追溯原因，运维必备；
- 异常容错
- 极简扩展：告警功能、检测逻辑、修复逻辑都能无缝扩展，不用动核心循环。
三、六大高频故障场景 + 专属解决方案（直接套用，全部可运行）
下面列举的 6个场景，是工作中90%的人都会遇到的高频问题，每个场景都基于上面的通用模板，给出「问题描述+解决方案+完整代码」，所有代码都经过实测，复制粘贴就能用，按需微调参数即可。
场景一：核心程序/服务 意外挂掉，自动重启（最常用！必配）
问题描述
- 你的核心业务程序（比如Python后端服务、爬虫脚本、数据同步程序）、Java服务、Node服务，因为内存溢出、代码bug、系统资源不足等原因，运行中突然崩溃退出；
- 定时任务脚本运行中报错终止，无人知晓，导致业务中断；
- 痛点：需要人工登录服务器重启，半夜出问题就得爬起来处理。
解决方案：进程存活看门狗
核心逻辑：检测目标进程是否存在 → 不存在则执行重启命令 → 重启成功即修复完成。
# ========== 场景一：进程存活看门狗（继承通用基类） ==========
classProcessWatchDog(BaseWatchDog):
defcheck_status(self):
"""检测：目标进程是否存活"""
process_name = self.target_config["process_name"]
# Windows用tasklist，Linux/Mac用ps，跨平台兼容
if os.name == "nt":
cmd = f"tasklist | findstr /i {process_name}"
else:
cmd = f"ps -ef | grep {process_name} | grep -v grep"
try:
result = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# 有输出说明进程存在
returnlen(result.stdout) > 0
except:
returnFalse
defrepair(self):
"""修复：自动重启目标服务/程序"""
service_cmd = self.target_config["service_cmd"]
try:
# 后台启动进程，不阻塞看门狗
if os.name == "nt":
subprocess.Popen(service_cmd, shell=True)
else:
subprocess.Popen(service_cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
time.sleep(3) # 等待3秒，让进程启动完成
# 再次检测，确认重启成功
returnself.check_status()
except Exception as e:
logger.error(f"重启服务失败：{str(e)}")
returnFalse
# 启动进程看门狗
if __name__ == "__main__":
# 按需修改配置：进程名+重启命令
TARGET_CONFIG = {
"process_name": "python",
"service_cmd": "python ./my_business_service.py"# 你的核心程序启动命令
}
watchdog = ProcessWatchDog()
watchdog.run()
场景二：配置文件被误改/篡改，自动恢复正确配置（开篇痛点解决！）
问题描述
- 这是开篇提到的核心痛点：线上服务的配置文件（
.ini/.yml/.json/.conf）被同事误改、被脚本覆盖、被恶意篡改，导致服务参数错误、数据库连接失败、接口调用异常； - 现象：服务运行中报错，但进程还在，日志里全是配置相关的错误；
- 痛点：半夜发现配置错了，必须手动登录服务器改回正确配置，改完还要重启服务，极其折腾。
解决方案：配置文件看门狗
核心逻辑：提前保存一份正确的配置文件备份 → 检测当前配置文件的MD5值是否和备份一致 → 不一致说明被修改 → 用备份文件覆盖当前配置 → 重启服务完成修复。
import hashlib
# ========== 场景二：配置文件防篡改看门狗（继承通用基类） ==========
classConfigFileWatchDog(BaseWatchDog):
def__init__(self):
super().__init__()
self.backup_config_path = "./config_backup.ini"# 正确的配置备份文件，务必提前保存
# 提前计算备份配置的MD5值
self.correct_md5 = self.get_file_md5(self.backup_config_path)
defget_file_md5(self, file_path):
"""计算文件MD5值，用于校验文件是否被修改"""
md5 = hashlib.md5()
withopen(file_path, "rb") as f:
while chunk := f.read(4096):
md5.update(chunk)
return md5.hexdigest()
defcheck_status(self):
"""检测：配置文件是否被修改"""
current_config_path = self.target_config["config_path"]
ifnot os.path.exists(current_config_path):
returnFalse
current_md5 = self.get_file_md5(current_config_path)
# MD5一致则未被修改，不一致则被篡改
return current_md5 == self.correct_md5
defrepair(self):
"""修复：用备份覆盖被修改的配置 + 重启服务"""
try:
# 第一步：覆盖错误配置，恢复正确配置
withopen(self.backup_config_path, "rb") as f1, open(self.target_config["config_path"], "wb") as f2:
f2.write(f1.read())
logger.info("配置文件已恢复为正确备份版本")
# 第二步：重启服务，让新配置生效
service_cmd = self.target_config["service_cmd"]
if os.name == "nt":
subprocess.Popen(service_cmd, shell=True)
else:
subprocess.Popen(service_cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
returnTrue
except Exception as e:
logger.error(f"恢复配置失败：{str(e)}")
returnFalse
# 启动配置文件看门狗
if __name__ == "__main__":
TARGET_CONFIG = {
"config_path": "./config.ini", # 被监控的配置文件
"service_cmd": "python ./my_business_service.py"# 服务重启命令
}
watchdog = ConfigFileWatchDog()
watchdog.run()
✨ 关键提醒：务必提前把正确的配置文件保存为备份文件，放在安全路径，不要被误删！
场景三：服务器磁盘空间爆满，自动清理日志/垃圾文件
问题描述
- 服务器的日志文件（比如
nginx.log、app.log）每天疯狂写入，没有自动清理机制，久而久之磁盘分区被占满； - 爬虫/程序生成的临时文件、缓存文件堆积，占用大量空间；
- 现象：服务写入文件失败、数据库无法写入、系统卡死，报错提示「No space left on device」；
- 痛点：磁盘满了之后，所有服务都会出问题，必须手动登录清理，而且容易遗漏垃圾文件。
解决方案：磁盘空间看门狗
核心逻辑：检测指定磁盘分区的使用率 → 超过阈值（比如85%） → 自动清理指定目录的日志/临时文件 → 清理完成后再次检测磁盘使用率，确认恢复正常。
import shutil
# ========== 场景三：磁盘空间看门狗（继承通用基类） ==========
classDiskSpaceWatchDog(BaseWatchDog):
def__init__(self):
super().__init__()
self.disk_path = "/"if os.name != "nt"else"C:"# 监控的磁盘分区
self.warning_threshold = 85# 磁盘使用率阈值(%)，超过则清理
self.clean_dirs = ["./logs", "./temp"] # 需要清理的日志/临时文件目录
defcheck_status(self):
"""检测：磁盘使用率是否超过阈值"""
disk_usage = shutil.disk_usage(self.disk_path)
usage_percent = (disk_usage.used / disk_usage.total) * 100
logger.info(f"当前磁盘使用率：{usage_percent:.2f}%")
return usage_percent < self.warning_threshold
defrepair(self):
"""修复：自动清理日志和临时文件"""
try:
total_clean_size = 0
for dir_path inself.clean_dirs:
ifnot os.path.exists(dir_path):
continue
# 遍历目录下的所有文件，删除日志和临时文件
for root, dirs, files in os.walk(dir_path):
for file in files:
if file.endswith((".log", ".tmp", ".bak")):
file_path = os.path.join(root, file)
file_size = os.path.getsize(file_path)
os.remove(file_path)
total_clean_size += file_size
logger.info(f"清理完成，共释放空间：{total_clean_size / 1024 / 1024:.2f} MB")
# 再次检测磁盘使用率
returnself.check_status()
except Exception as e:
logger.error(f"清理磁盘失败：{str(e)}")
returnFalse
# 启动磁盘空间看门狗
if __name__ == "__main__":
watchdog = DiskSpaceWatchDog()
watchdog.run()
场景四：端口被占用/服务端口无监听，自动释放+重启
问题描述
- 你的服务需要监听指定端口（比如8080、5000），但因为服务异常退出、进程残留，端口被占用，新服务启动失败；
- 服务进程还在，但端口监听失效，客户端无法连接，出现「连接拒绝」错误；
- 痛点：需要手动执行
netstat/lsof命令找到占用端口的进程，kill掉之后再重启服务，步骤繁琐。
解决方案：端口监听看门狗
核心逻辑：检测指定端口是否有进程监听 → 无监听/被占用 → 自动找到占用端口的进程并杀死 → 重启服务，重新监听端口。
import socket
# ========== 场景四：端口监听看门狗（继承通用基类） ==========
classPortWatchDog(BaseWatchDog):
def__init__(self):
super().__init__()
self.target_port = 8080# 需要监控的端口
defcheck_status(self):
"""检测：指定端口是否有进程监听"""
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(1)
try:
# 连接本机端口，能连接说明端口正常监听
result = s.connect_ex(("127.0.0.1", self.target_port))
return result == 0
finally:
s.close()
defkill_port_occupier(self):
"""杀死占用指定端口的进程"""
port = self.target_port
if os.name == "nt":
# Windows 命令
cmd = f"netstat -ano | findstr :{port}"
else:
# Linux/Mac 命令
cmd = f"lsof -i :{port} | grep LISTEN | awk '{{print $2}}'"
try:
result = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
pids = result.stdout.decode().strip().split()
for pid in pids:
if pid.isdigit():
if os.name == "nt":
subprocess.run(f"taskkill /F /PID {pid}", shell=True)
else:
subprocess.run(f"kill -9 {pid}", shell=True)
logger.info(f"已杀死占用端口{port}的进程，PID：{pid}")
returnTrue
except Exception as e:
logger.error(f"释放端口失败：{str(e)}")
returnFalse
defrepair(self):
"""修复：释放端口 + 重启服务"""
# 第一步：杀死占用端口的进程
self.kill_port_occupier()
time.sleep(2)
# 第二步：重启服务
service_cmd = self.target_config["service_cmd"]
try:
subprocess.Popen(service_cmd, shell=True)
time.sleep(3)
returnself.check_status()
except Exception as e:
logger.error(f"重启服务失败：{str(e)}")
returnFalse
# 启动端口看门狗
if __name__ == "__main__":
TARGET_CONFIG = {
"service_cmd": "python ./server.py"# 你的服务启动命令
}
watchdog = PortWatchDog()
watchdog.run()
场景五：数据库/第三方服务连接失败，自动重连+告警
问题描述
- 你的程序需要连接MySQL/Redis/MongoDB，或者调用第三方API接口，因为网络波动、数据库服务重启、账号密码过期等原因，连接失败；
- 现象：程序日志里全是「连接超时」「认证失败」错误，业务无法读写数据；
- 痛点：网络恢复后，程序不会自动重连，需要手动重启服务，而且如果是账号密码问题，需要及时告警。
解决方案：服务连通性看门狗
核心逻辑：定时发起数据库/第三方服务的连接测试 → 连接失败则执行重连逻辑 → 多次重连失败则发送紧急告警，提醒人工排查。
import pymysql # 需安装：pip install pymysql
import redis # 需安装：pip install redis
# ========== 场景五：数据库/服务连通性看门狗（继承通用基类） ==========
classConnectivityWatchDog(BaseWatchDog):
def__init__(self):
super().__init__()
# 数据库/Redis配置
self.db_config = {
"host": "127.0.0.1",
"port": 3306,
"user": "root",
"password": "123456",
"database": "test_db"
}
self.redis_config = {"host": "127.0.0.1", "port": 6379, "db": 0}
self.retry_times = 3# 重连次数
defcheck_status(self):
"""检测：数据库+Redis是否能正常连接"""
# 检测MySQL
try:
db_conn = pymysql.connect(**self.db_config)
db_conn.close()
except:
logger.error("MySQL数据库连接失败")
returnFalse
# 检测Redis
try:
redis_conn = redis.Redis(** self.redis_config)
redis_conn.ping()
except:
logger.error("Redis连接失败")
returnFalse
returnTrue
defrepair(self):
"""修复：多次重连，失败则告警"""
for i inrange(1, self.retry_times+1):
logger.info(f"第{i}次尝试重连服务...")
ifself.check_status():
returnTrue
time.sleep(5)
# 多次重连失败，返回False，触发紧急告警
returnFalse
# 启动连通性看门狗
if __name__ == "__main__":
watchdog = ConnectivityWatchDog()
watchdog.run()
场景六：日志文件出现关键错误，自动告警+备份日志
问题描述
- 你的程序运行中出现致命错误（比如KeyError、IndexError、数据库死锁），这些错误会写入日志文件，但进程不会退出，服务看似正常运行，实则业务逻辑出错；
- 现象：日志里有大量「ERROR」「CRITICAL」关键字，用户反馈功能异常，但进程还在；
- 痛点：无法实时发现日志中的错误，等用户反馈时已经造成损失。
解决方案：日志异常看门狗
核心逻辑：定时读取日志文件的最新内容 → 检测是否包含指定的错误关键字 → 发现错误则立即发送告警 → 自动备份错误日志，方便后续排查。
# ========== 场景六：日志异常监控看门狗（继承通用基类） ==========
classLogErrorWatchDog(BaseWatchDog):
def__init__(self):
super().__init__()
self.log_path = "./app.log"# 被监控的日志文件
self.error_keywords = ["ERROR", "CRITICAL", "KeyError", "TimeoutError", "ConnectionRefusedError"]
self.last_read_pos = 0# 记录上次读取的日志位置，只看新日志，提升效率
defcheck_status(self):
"""检测：日志中是否出现关键错误关键字"""
ifnot os.path.exists(self.log_path):
returnTrue
has_error = False
withopen(self.log_path, "r", encoding="utf-8") as f:
f.seek(self.last_read_pos) # 跳到上次读取的位置
new_content = f.read()
self.last_read_pos = f.tell() # 更新读取位置
for keyword inself.error_keywords:
if keyword in new_content:
has_error = True
logger.error(f"日志中发现错误关键字：{keyword}")
break
returnnot has_error
defrepair(self):
"""修复：备份错误日志 + 发送告警"""
try:
# 备份当前日志到新文件，带时间戳
backup_log = f"./error_log_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
shutil.copy(self.log_path, backup_log)
logger.info(f"错误日志已备份至：{backup_log}")
# 这里可以扩展：清空日志文件，防止日志过大
# with open(self.log_path, "w") as f: f.write("")
returnTrue
except Exception as e:
logger.error(f"备份日志失败：{str(e)}")
returnFalse
# 启动日志异常看门狗
if __name__ == "__main__":
watchdog = LogErrorWatchDog()
watchdog.run()
四、进阶优化：让你的看门狗脚本更健壮（必做的6个优化点）
上面的6个场景脚本已经能解决绝大多数问题，在此基础上，做一些简单的优化，就能让看门狗从「能用」变成「好用」，稳定性拉满，真正做到「无人值守」。所有优化点都简单易实现，建议全部加上：
优化1：扩展告警方式（重中之重）
基础版的告警只是打印日志，强烈建议扩展为 「钉钉/企业微信机器人告警」「邮件告警」，这样即使不在电脑前，手机也能收到告警通知，做到「有问题及时知会」。
附：Python调用钉钉机器人只需几行代码，复制官方文档的示例即可，无任何门槛。
优化2：后台常驻运行，不怕终端关闭
- Linux/Mac：用
nohup python watchdog.py & 命令启动，看门狗会在后台常驻运行，关闭终端也不会停止； - Windows：创建一个批处理文件，用
pythonw.exe watchdog.py 启动，无窗口运行，开机自启可添加到系统服务。
优化3：增加「连续异常」判断，避免频繁修复
比如：磁盘使用率超过阈值，第一次检测到不立即清理，连续3次检测都超过阈值再执行清理，避免因为临时文件导致的误操作。
优化4：看门狗自身的进程守护
可以给看门狗脚本也加一层守护：用supervisor（Linux）或winsw（Windows）监控看门狗进程，防止看门狗自己挂掉，做到「双重保险」。
优化5：配置参数外置
把所有配置（检测间隔、阈值、文件路径）写到独立的config.yml文件中，不用改代码就能调整参数，更灵活。
优化6：多场景合并监控
不用为每个场景单独启动一个看门狗，可在一个脚本中集成多个检测逻辑，比如「进程存活+端口监听+磁盘空间」一起监控，减少进程数量，方便管理。
五、文末总结：为什么这个看门狗脚本值得你用？
- 解决痛点直击核心：所有场景都是工作中高频的「半夜爬起来处理」的问题，脚本的核心就是用自动化替代人工，让机器替人干活；
- 极简易用，无门槛：所有代码都是Python原生语法，无需部署复杂框架，复制粘贴就能用，零基础也能看懂和修改；
- 通用性极强：6个场景覆盖了开发/运维的绝大多数需求，而且基于通用基类，新增场景只需几行代码，扩展性拉满；
- 轻量化无负担：脚本占用的CPU和内存极低，运行在服务器上完全不会影响业务程序，是「零成本提升运维效率」的利器。
最后想说的话
我们写代码的初衷，从来都不是为了炫技，而是为了解决问题、解放自己。一个简单的看门狗脚本，看似不起眼，却能让你告别半夜被叫醒的痛苦，让你拥有安稳的睡眠，让你的服务更稳定、更可靠。
从此，半夜不用爬起来改配置、不用远程重启服务、不用人工巡检——你的服务，交给这个忠诚的「看门狗」就够了。
希望这篇文章能帮到你，愿你从此告别运维焦虑，专注于更有价值的事情 ✨。
