Linux 下实现秒级定时任务的 5 种方法详解

在实际开发中，我们经常需要执行定时任务。虽然 crontab 是 Linux 下最常用的定时任务工具，但它最小只能设置到分钟级别。本文将详细介绍 5 种实现秒级定时任务的方案，并附上完整代码示例。

1. 问题背景：为什么需要秒级定时任务？

1.1 常见应用场景

实时数据监控和采集
高频数据更新
实时心跳检测
秒级轮询任务
高频数据同步

1.2 crontab 的局限性

1# crontab 最小时间单位是分钟
2* * * * * command    # 每分钟执行
3*/1 * * * * command  # 同样是每分钟执行
4

2. 解决方案一览

方案	适用场景	复杂度	可靠性
sleep 循环	简单秒级任务	⭐⭐	高
while 无限循环	长期运行任务	⭐⭐	高
systemd 定时器	系统级服务	⭐⭐⭐⭐	很高
Python 内置调度	Python 项目	⭐⭐⭐	高
schedule 库	复杂调度需求	⭐⭐⭐	高

3. 方案详解与代码实现

3.1 方案一：使用 sleep 命令循环执行（推荐）

原理：利用 crontab 每分钟触发，在脚本内使用循环实现秒级执行。

实现步骤：

创建主执行脚本 run_every_second.sh：

1#!/bin/bash
2# 描述：每秒执行 Python 脚本的包装脚本
3# 作者：你的名字
4# 日期：2024-01-20
5
6SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
7PYTHON_SCRIPT="$SCRIPT_DIR/get.py"
8LOG_FILE="$SCRIPT_DIR/execution.log"
9
10# 检查 Python 脚本是否存在
11if [ ! -f "$PYTHON_SCRIPT" ]; then
12    echo "$(date): 错误 - Python 脚本不存在: $PYTHON_SCRIPT" >> "$LOG_FILE"
13    exit 1
14fi
15
16# 循环执行 60 次（1分钟）
17for i in {1..60}; do
18    echo "$(date): 第 $i 次执行开始" >> "$LOG_FILE"
19    
20    # 执行 Python 脚本并记录输出
21    /usr/bin/python3 "$PYTHON_SCRIPT" >> "$LOG_FILE" 2>&1
22    
23    # 检查执行结果
24    if [ $? -eq 0 ]; then
25        echo "$(date): 第 $i 次执行成功" >> "$LOG_FILE"
26    else
27        echo "$(date): 第 $i 次执行失败" >> "$LOG_FILE"
28    fi
29    
30    # 等待 1 秒
31    sleep 1
32done
33
34echo "$(date): 分钟任务周期结束" >> "$LOG_FILE"
35

设置执行权限：

1chmod +x /path/to/run_every_second.sh
2

配置 crontab：

1crontab -e
2
3# 添加以下内容
4* * * * * /home/username/your_path/run_every_second.sh
5

优缺点分析：

✅ 优点：简单可靠，易于调试，依赖少
❌ 缺点：依赖 crontab，每分钟重启一次进程

3.2 方案二：使用 while 无限循环

原理：创建守护进程，使用无限循环实现持续执行。

实现代码：

创建守护脚本 daemon_runner.sh：

1#!/bin/bash
2# 描述：秒级任务守护进程
3# 使用说明：nohup ./daemon_runner.sh &
4
5SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
6PYTHON_SCRIPT="$SCRIPT_DIR/get.py"
7LOG_FILE="$SCRIPT_DIR/daemon.log"
8PID_FILE="$SCRIPT_DIR/daemon.pid"
9
10# 记录 PID
11echo $$ > "$PID_FILE"
12
13# 信号处理函数
14cleanup() {
15    echo "$(date): 接收到退出信号，清理资源..." >> "$LOG_FILE"
16    rm -f "$PID_FILE"
17    exit 0
18}
19
20trap cleanup SIGTERM SIGINT
21
22echo "$(date): 守护进程启动，PID: $$" >> "$LOG_FILE"
23
24# 主循环
25while true; do
26    START_TIME=$(date +%s)
27    
28    echo "$(date): 开始执行任务" >> "$LOG_FILE"
29    
30    # 执行 Python 脚本
31    if /usr/bin/python3 "$PYTHON_SCRIPT" >> "$LOG_FILE" 2>&1; then
32        echo "$(date): 任务执行成功" >> "$LOG_FILE"
33    else
34        echo "$(date): 任务执行失败" >> "$LOG_FILE"
35    fi
36    
37    END_TIME=$(date +%s)
38    EXECUTION_TIME=$((END_TIME - START_TIME))
39    
40    # 计算需要睡眠的时间（确保每秒执行）
41    if [ $EXECUTION_TIME -lt 1 ]; then
42        SLEEP_TIME=1
43    else
44        SLEEP_TIME=1
45    fi
46    
47    sleep $SLEEP_TIME
48done
49

启动和停止脚本 manage_daemon.sh：

1#!/bin/bash
2
3SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
4DAEMON_SCRIPT="$SCRIPT_DIR/daemon_runner.sh"
5PID_FILE="$SCRIPT_DIR/daemon.pid"
6
7case "$1" in
8    start)
9        if [ -f "$PID_FILE" ]; then
10            PID=$(cat "$PID_FILE")
11            if kill -0 "$PID" 2>/dev/null; then
12                echo "守护进程已在运行 (PID: $PID)"
13                exit 1
14            else
15                rm -f "$PID_FILE"
16            fi
17        fi
18        
19        nohup "$DAEMON_SCRIPT" > /dev/null 2>&1 &
20        echo "守护进程启动成功"
21        ;;
22    stop)
23        if [ -f "$PID_FILE" ]; then
24            PID=$(cat "$PID_FILE")
25            kill "$PID" 2>/dev/null
26            rm -f "$PID_FILE"
27            echo "守护进程已停止"
28        else
29            echo "守护进程未运行"
30        fi
31        ;;
32    status)
33        if [ -f "$PID_FILE" ]; then
34            PID=$(cat "$PID_FILE")
35            if kill -0 "$PID" 2>/dev/null; then
36                echo "守护进程运行中 (PID: $PID)"
37            else
38                echo "PID 文件存在但进程未运行"
39                rm -f "$PID_FILE"
40            fi
41        else
42            echo "守护进程未运行"
43        fi
44        ;;
45    *)
46        echo "用法: $0 {start|stop|status}"
47        exit 1
48        ;;
49esac
50

使用方法：

1# 启动
2./manage_daemon.sh start
3
4# 停止
5./manage_daemon.sh stop
6
7# 查看状态
8./manage_daemon.sh status
9

3.3 方案三：使用 systemd 定时器（系统级方案）

原理：利用 systemd 的高精度定时器实现秒级调度。

实现步骤：

创建 service 文件 /etc/systemd/system/seconds-task.service：

1[Unit]
2Description=Run Python script every second
3After=network.target
4
5[Service]
6Type=simple
7User=your_username
8WorkingDirectory=/home/your_username/your_path
9ExecStart=/usr/bin/python3 /home/your_username/your_path/get.py
10Restart=always
11RestartSec=1
12
13# 日志配置
14StandardOutput=journal
15StandardError=journal
16
17[Install]
18WantedBy=multi-user.target
19

创建 timer 文件 /etc/systemd/system/seconds-task.timer：

1[Unit]
2Description=Run Python script every second
3Requires=seconds-task.service
4
5[Timer]
6# 系统启动后 1 秒开始
7OnBootSec=1s
8# 每次服务激活后 1 秒再次执行
9OnUnitActiveSec=1s
10# 定时器精度（尽可能精确）
11AccuracySec=1ms
12# 如果错过执行时间，立即执行
13Persistent=true
14
15[Install]
16WantedBy=timers.target
17

启用并启动服务：

1# 重新加载 systemd 配置
2sudo systemctl daemon-reload
3
4# 启用定时器
5sudo systemctl enable seconds-task.timer
6
7# 启动定时器
8sudo systemctl start seconds-task.timer
9
10# 查看状态
11sudo systemctl status seconds-task.timer
12sudo systemctl status seconds-task.service
13
14# 查看日志
15sudo journalctl -u seconds-task.service -f
16

3.4 方案四：Python 内置调度

原理：在 Python 脚本内部实现定时循环。

完整代码示例：

1#!/usr/bin/env python3
2"""
3秒级定时任务 - Python 内置调度方案
4功能：每秒执行一次数据采集任务
5作者：你的名字
6日期：2024-01-20
7"""
8
9import time
10import signal
11import sys
12import logging
13from datetime import datetime
14import requests
15from bs4 import BeautifulSoup
16
17# 配置日志
18logging.basicConfig(
19    level=logging.INFO,
20    format='%(asctime)s - %(levelname)s - %(message)s',
21    handlers=[
22        logging.FileHandler('scheduler.log', encoding='utf-8'),
23        logging.StreamHandler()
24    ]
25)
26
27logger = logging.getLogger(__name__)
28
29class SecondScheduler:
30    def __init__(self, interval=1):
31        """
32        初始化调度器
33        
34        Args:
35            interval: 执行间隔（秒）
36        """
37        self.interval = interval
38        self.running = False
39        self.iteration = 0
40        
41        # 注册信号处理
42        signal.signal(signal.SIGINT, self.signal_handler)
43        signal.signal(signal.SIGTERM, self.signal_handler)
44    
45    def signal_handler(self, signum, frame):
46        """信号处理函数"""
47        logger.info(f"接收到信号 {signum}，准备退出...")
48        self.running = False
49    
50    def main_task(self):
51        """
52        主要任务逻辑
53        这里替换为你的实际业务代码
54        """
55        try:
56            self.iteration += 1
57            current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
58            
59            logger.info(f"第 {self.iteration} 次执行 - 时间: {current_time}")
60            
61            # 示例：模拟你的数据采集任务
62            # 这里替换为你的实际业务逻辑
63            result = self.sample_data_collection()
64            
65            logger.info(f"任务执行完成，结果: {result}")
66            return True
67            
68        except Exception as e:
69            logger.error(f"任务执行失败: {e}")
70            return False
71    
72    def sample_data_collection(self):
73        """
74        示例数据采集函数
75        替换为你的实际数据采集逻辑
76        """
77        # 这里是你原来的爬虫代码
78        # 示例返回
79        return {"status": "success", "data": "sample data"}
80    
81    def run(self):
82        """启动调度器"""
83        logger.info("秒级调度器启动")
84        self.running = True
85        
86        try:
87            while self.running:
88                start_time = time.time()
89                
90                # 执行主要任务
91                self.main_task()
92                
93                # 计算执行时间并调整睡眠时间
94                execution_time = time.time() - start_time
95                sleep_time = max(0, self.interval - execution_time)
96                
97                # 精确睡眠
98                if sleep_time > 0:
99                    time.sleep(sleep_time)
100                else:
101                    logger.warning(f"任务执行时间 {execution_time:.3f}s 超过间隔 {self.interval}s")
102                    
103        except KeyboardInterrupt:
104            logger.info("用户中断执行")
105        except Exception as e:
106            logger.error(f"调度器异常: {e}")
107        finally:
108            logger.info(f"调度器停止，共执行 {self.iteration} 次")
109    
110    def run_with_retry(self, max_retries=3):
111        """
112        带重试机制的运行方法
113        
114        Args:
115            max_retries: 最大重试次数
116        """
117        retries = 0
118        while retries < max_retries:
119            try:
120                self.run()
121                break
122            except Exception as e:
123                retries += 1
124                logger.error(f"第 {retries} 次运行失败: {e}")
125                if retries < max_retries:
126                    logger.info(f"{60} 秒后重试...")
127                    time.sleep(60)
128                else:
129                    logger.error(f"达到最大重试次数 {max_retries}，停止运行")
130
131if __name__ == "__main__":
132    # 创建调度器实例（每秒执行）
133    scheduler = SecondScheduler(interval=1)
134    
135    # 启动调度器（带重试机制）
136    scheduler.run_with_retry(max_retries=5)
137

3.5 方案五：使用 Python schedule 库

原理：使用第三方库提供更友好的调度接口。

实现代码：

1#!/usr/bin/env python3
2"""
3秒级定时任务 - schedule 库方案
4依赖：pip install schedule
5"""
6
7import schedule
8import time
9import logging
10from datetime import datetime
11
12# 配置日志
13logging.basicConfig(
14    level=logging.INFO,
15    format='%(asctime)s - %(levelname)s - %(message)s'
16)
17logger = logging.getLogger(__name__)
18
19class ScheduleRunner:
20    def __init__(self):
21        self.running = False
22        
23    def task_function(self):
24        """需要执行的任务函数"""
25        try:
26            current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
27            logger.info(f"执行任务 - 时间: {current_time}")
28            
29            # 这里替换为你的实际任务逻辑
30            # 你的数据采集代码...
31            
32            return True
33            
34        except Exception as e:
35            logger.error(f"任务执行异常: {e}")
36            return False
37    
38    def setup_schedule(self):
39        """设置调度规则"""
40        # 每秒执行
41        schedule.every(1).seconds.do(self.task_function)
42        
43        # 也可以设置更复杂的规则
44        # schedule.every(5).seconds.do(self.task_function)  # 每5秒
45        # schedule.every().minute.at(":00").do(self.task_function)  # 每分钟的0秒
46    
47    def run(self):
48        """启动调度器"""
49        logger.info("schedule 调度器启动")
50        self.running = True
51        self.setup_schedule()
52        
53        try:
54            while self.running:
55                # 运行所有到期的任务
56                schedule.run_pending()
57                
58                # 短暂睡眠以减少 CPU 使用
59                time.sleep(0.1)
60                
61        except KeyboardInterrupt:
62            logger.info("用户中断执行")
63        except Exception as e:
64            logger.error(f"调度器异常: {e}")
65        finally:
66            logger.info("调度器停止")
67    
68    def stop(self):
69        """停止调度器"""
70        self.running = False
71
72if __name__ == "__main__":
73    runner = ScheduleRunner()
74    runner.run()
75

4. 实战建议与注意事项

4.1 选择合适的方案

需求场景	推荐方案	理由
简单秒级任务	方案一（sleep循环）	简单可靠，易于维护
长期运行服务	方案二（while循环）	稳定性好，可控性强
系统级服务	方案三（systemd）	系统集成度高，管理方便
Python 项目	方案四/五	纯 Python 实现，与项目集成度高

4.2 性能优化建议

避免任务执行时间超过间隔：

1# 在任务中检查执行时间
2start_time = time.time()
3# 执行任务...
4execution_time = time.time() - start_time
5if execution_time > interval:
6    logger.warning(f"任务执行时间 {execution_time} 超过间隔 {interval}")
7

添加异常处理：

1try:
2    # 任务代码
3except Exception as e:
4    logger.error(f"任务执行失败: {e}")
5    # 根据需求决定是否继续执行
6

资源管理：

1# 定期清理资源，防止内存泄漏
2if self.iteration % 1000 == 0:
3    self.cleanup_resources()
4

4.3 监控与日志

为每个方案添加完善的日志：

1import logging
2
3logging.basicConfig(
4    level=logging.INFO,
5    format='%(asctime)s - %(levelname)s - %(message)s',
6    handlers=[
7        logging.FileHandler('task.log'),
8        logging.StreamHandler()
9    ]
10)
11

5. 总结

本文详细介绍了 5 种在 Linux 下实现秒级定时任务的方法，每种方法都有其适用场景：

方案一 适合简单场景，依赖少
方案二 适合长期运行的服务
方案三 适合系统级服务部署
方案四 适合 Python 项目集成
方案五 适合复杂的调度需求

在实际项目中，建议根据具体需求选择合适的方案，并添加完善的错误处理、日志记录和监控机制，确保任务的稳定运行。

温馨提示：高频请求请注意遵守目标网站的 robots.txt 和服务条款，避免对目标服务器造成过大压力。

《Linux 下实现秒级定时任务的 5 种方法详解》是转载文章，点击查看原文。