Python如何实现任务入队调度、按指定时间执行带参数POST请求脚本

以下是两种不同复杂度的实现思路，均比多进程+sleep的方式更节省资源、易维护：

轻量单机方案（无需额外第三方依赖） #

用Python标准库sched事件调度器 + 本地JSON文件做任务存储，仅需两个脚本即可完成需求：

1. 任务提交脚本 add_task.py #

作用是接收ID和执行时间参数，将任务写入本地任务池文件，无需常驻运行：

import json
import sys
from datetime import datetime

# 接收参数：python add_task.py [任务ID] [执行时间 格式%Y-%m-%d %H:%M:%S]
task_id = sys.argv[1]
exec_time = sys.argv[2]
# 校验时间格式合法性
datetime.strptime(exec_time, "%Y-%m-%d %H:%M:%S")

# 追加写入任务池
with open("task_pool.json", "r+", encoding="utf-8") as f:
    try:
        tasks = json.load(f)
    except json.JSONDecodeError:
        tasks = []
    tasks.append({"id": task_id, "exec_time": exec_time, "status": "pending"})
    f.seek(0)
    json.dump(tasks, f, indent=2, ensure_ascii=False)

2. 任务执行常驻脚本 worker.py #

后台常驻运行，自动加载任务、计算触发时间、到点执行POST请求：

import sched
import time
import json
import requests
from datetime import datetime, timedelta

# 初始化调度器
scheduler = sched.scheduler(time.time, time.sleep)
TASK_FILE = "task_pool.json"

# 核心POST执行逻辑
def run_post_task(task_id):
    print(f"执行任务ID: {task_id}，时间：{datetime.now()}")
    # 这里替换为你自己的POST请求逻辑
    # requests.post("你的接口地址", json={"id": task_id})

def load_tasks():
    """加载未执行的任务加入调度队列"""
    try:
        with open(TASK_FILE, "r", encoding="utf-8") as f:
            tasks = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        tasks = []
    now = time.time()
    for task in tasks:
        if task["status"] != "pending":
            continue
        exec_timestamp = datetime.strptime(task["exec_time"], "%Y-%m-%d %H:%M:%S").timestamp()
        if exec_timestamp < now:
            # 已过执行时间的直接标记为过期，可根据需求调整为立即执行
            task["status"] = "expired"
            continue
        # 计算延迟时间加入调度
        delay = exec_timestamp - now
        scheduler.enter(delay, 1, run_post_task, argument=(task["id"],))
        task["status"] = "scheduled"
    # 回写任务状态
    with open(TASK_FILE, "w", encoding="utf-8") as f:
        json.dump(tasks, f, indent=2, ensure_ascii=False)

if __name__ == "__main__":
    # 首次加载任务
    load_tasks()
    # 每10分钟刷新一次任务池，支持中途新增任务无需重启worker
    while True:
        scheduler.run(blocking=False)
        time.sleep(60)
        load_tasks()

该方案优势：仅需启动1个常驻worker进程即可管理所有任务，资源占用极低，适合任务量不大的单机使用场景。

进阶方案（支持持久化、任务重试、分布式部署） #

如果需要更高的可靠性，防止进程崩溃丢失任务，可使用APScheduler库 + SQLite做持久化存储：

1. 先安装依赖：pip install apscheduler
2. 同一个脚本即可支持任务添加和执行，示例代码：

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
import requests
from datetime import datetime
import sys

# 核心POST执行逻辑
def run_post_task(task_id):
    print(f"执行任务ID: {task_id}")
    # requests.post("你的接口地址", json={"id": task_id})

if __name__ == "__main__":
    # 初始化调度器，用SQLite持久化存储任务，重启不会丢失
    job_stores = {
        'default': SQLAlchemyJobStore(url='sqlite:///jobs.db')
    }
    scheduler = BackgroundScheduler(jobstores=job_stores)
    
    if len(sys.argv) > 2:
        # 传入参数时为添加任务模式：python scheduler.py [任务ID] [执行时间%Y-%m-%d %H:%M:%S]
        task_id = sys.argv[1]
        exec_time = datetime.strptime(sys.argv[2], "%Y-%m-%d %H:%M:%S")
        # 添加定时任务
        scheduler.add_job(
            run_post_task,
            'date',
            run_date=exec_time,
            args=[task_id],
            id=f"task_{task_id}_{exec_time.timestamp()}",
            replace_existing=True
        )
        print(f"任务{task_id}已添加，执行时间：{exec_time}")
    else:
        # 无参数时为执行worker模式
        scheduler.start()
        print("任务调度器已启动，等待执行任务...")
        # 保持进程运行
        while True:
            time.sleep(3600)

该方案优势：内置任务持久化、异常重试、执行日志等能力，支持动态增删改查任务，不需要自己维护任务存储逻辑，适合生产环境使用。

内容的提问来源于stack exchange，提问作者jack87