本文目录导读：
脚本的暂停与恢复有多种实现方式,具体取决于你的使用场景（是系统进程管理、定时任务，还是代码内部的流程控制），下面按常见场景分类说明：
操作系统级别（进程暂停/恢复）
Linux / Unix 使用信号
# 暂停进程（发送 SIGSTOP 信号） kill -STOP <PID> # 恢复进程（发送 SIGCONT 信号） kill -CONT <PID> # 示例：暂停当前脚本自身 trap 'echo "Paused"; kill -STOP $$; echo "Resumed"' SIGTSTP
Windows 使用 PowerShell
# 暂停进程 Suspend-Process -Name "notepad" # 恢复进程 Resume-Process -Name "notepad"
Python 脚本内部暂停/恢复
使用 threading.Event（推荐）
import threading
import time
class PauseableTask:
def __init__(self):
self.pause_event = threading.Event()
self.pause_event.set() # 初始为运行状态
def pause(self):
self.pause_event.clear()
def resume(self):
self.pause_event.set()
def run(self):
while True:
self.pause_event.wait() # 暂停时会阻塞在这里
# 执行实际任务
print("Working...")
time.sleep(1)
使用信号处理
import signal
import time
paused = False
def handler(signum, frame):
global paused
paused = not paused
if paused:
print("Paused")
else:
print("Resumed")
signal.signal(signal.SIGTSTP, handler)
while True:
while paused:
time.sleep(0.1)
# 实际任务代码
print("Running...")
time.sleep(1)
Bash 脚本暂停/恢复
简单方式
#!/bin/bash
trap 'echo "Paused"; kill -STOP $$; echo "Resumed"' SIGTSTP
while true; do
echo "Running task..."
sleep 1
done
使用文件锁控制
#!/bin/bash
PAUSE_FILE="/tmp/script_pause"
while true; do
# 检查暂停文件是否存在
if [ -f "$PAUSE_FILE" ]; then
echo "Paused, waiting..."
sleep 1
continue
fi
# 实际任务
echo "Working..."
sleep 1
done
定时任务（Cron）暂停/恢复
禁用/启用 Cron 任务
# 暂停（注释掉任务） crontab -e # 在任务前加 # 注释 # 恢复（去掉注释） crontab -e # 删除 #
使用控制脚本
#!/bin/bash
# control.sh
case "$1" in
pause)
touch /tmp/cron_paused
;;
resume)
rm -f /tmp/cron_paused
;;
status)
if [ -f /tmp/cron_paused ]; then
echo "Paused"
else
echo "Running"
fi
;;
esac
分布式任务队列（如 Celery）
# 暂停特定任务队列
celery control cancel_consumer -d my_queue
# 恢复
celery control add_consumer my_queue
# 或通过代码
from celery import current_app
current_app.control.cancel_consumer('my_queue')
current_app.control.add_consumer('my_queue')
综合示例：带状态监控的脚本
#!/usr/bin/env python3
import json
import time
import os
class TaskManager:
def __init__(self, status_file="task_status.json"):
self.status_file = status_file
self.init_status()
def init_status(self):
if not os.path.exists(self.status_file):
self.set_status("running")
def set_status(self, status):
with open(self.status_file, 'w') as f:
json.dump({"status": status, "timestamp": time.time()}, f)
def get_status(self):
try:
with open(self.status_file, 'r') as f:
return json.load(f)["status"]
except:
return "unknown"
def pause(self):
self.set_status("paused")
print("Task paused")
def resume(self):
self.set_status("running")
print("Task resumed")
def run(self):
while True:
status = self.get_status()
if status == "paused":
time.sleep(1)
continue
elif status == "stopped":
print("Task stopped")
break
# 执行任务
print(f"Running task at {time.time()}")
time.sleep(2)
if __name__ == "__main__":
tm = TaskManager()
# 可以通过外部命令控制
# python script.py pause
# python script.py resume
import sys
if len(sys.argv) > 1:
cmd = sys.argv[1]
if cmd == "pause":
tm.pause()
elif cmd == "resume":
tm.resume()
elif cmd == "stop":
tm.set_status("stopped")
else:
tm.run()
最佳实践建议
-
选择适当的方法：
- 系统进程：使用信号
- 单脚本：使用 Event 或全局变量
- 分布式系统：使用任务队列的控制接口
-
状态持久化：如果需要重启后保持状态，使用文件或数据库
-
优雅处理：暂停时确保当前任务完成，避免数据不一致
-
监控与日志：记录暂停/恢复的时间点和原因
根据你的具体场景（是控制单个脚本、整个系统进程，还是分布式任务），选择最合适的实现方式。
