基于用户间隔触发的Post请求发布服务最优实现方案咨询

你的核心问题是频繁查询数据库检查任务状态导致效率低下，同时需要实现「按用户指定间隔触发POST请求+可终止任务」的需求，以下是从简到繁的几种方案，可根据你的场景选择：

方案1：优化原逻辑，减少数据库查询（无额外依赖，快速见效） #

不用引入新组件，直接降低数据库查询频率，改用内存缓存任务状态（单实例场景），同时提供接口主动更新状态：

from fastapi import FastAPI, BackgroundTasks
import time
import requests
import threading
from typing import Dict

app = FastAPI()
# 单实例用内存存储任务状态，多实例需替换为Redis
task_status: Dict[str, str] = {}
task_events: Dict[str, threading.Event] = {}

def scheduled_post(interval: int, request_no: str, total_duration: int, url: str):
    end_time = time.time() + total_duration
    event = task_events[request_no]
    
    while time.time() < end_time:
        # 直接读取内存状态，无需查库
        if task_status.get(request_no) == "stop":
            break
        # 发送POST请求
        requests.post(url)
        # 等待设定间隔，支持被stop信号中断
        if event.wait(interval):
            break
    
    # 清理任务状态
    task_status.pop(request_no, None)
    task_events.pop(request_no, None)

@app.post("/start-task")
async def start_task(interval: int, request_no: str, total_duration: int, url: str, background_tasks: BackgroundTasks):
    task_status[request_no] = "running"
    task_events[request_no] = threading.Event()
    background_tasks.add_task(scheduled_post, interval, request_no, total_duration, url)
    return {"message": "Task started"}

@app.post("/stop-task")
async def stop_task(request_no: str):
    task_status[request_no] = "stop"
    event = task_events.get(request_no)
    if event:
        event.set()
    return {"message": "Task stopped"}

• 优点：零额外依赖，实现简单，性能提升明显；
• 缺点：仅支持单实例部署，多实例下内存状态无法共享。

方案2：FastAPI + Redis（支持多实例，轻量高效） #

如果需要多实例部署，将内存状态替换为Redis，同时用Redis Pub/Sub实现实时任务终止通知（无需轮询）：

from fastapi import FastAPI, BackgroundTasks
import time
import requests
import redis

app = FastAPI()
# 初始化Redis连接
r = redis.Redis(host="localhost", port=6379, db=0)

def scheduled_post(interval: int, request_no: str, total_duration: int, url: str):
    pubsub = r.pubsub()
    pubsub.subscribe(f"task_stop:{request_no}")
    end_time = time.time() + total_duration

    while time.time() < end_time:
        # 监听stop信号，无需轮询数据库
        message = pubsub.get_message(ignore_subscribe_messages=True)
        if message and message["data"].decode() == "stop":
            break
        # 发送POST请求
        requests.post(url)
        time.sleep(interval)
    
    pubsub.close()
    # 清理Redis状态
    r.delete(f"task_status:{request_no}")

@app.post("/start-task")
async def start_task(interval: int, request_no: str, total_duration: int, url: str, background_tasks: BackgroundTasks):
    r.set(f"task_status:{request_no}", "running")
    background_tasks.add_task(scheduled_post, interval, request_no, total_duration, url)
    return {"message": "Task started"}

@app.post("/stop-task")
async def stop_task(request_no: str):
    r.publish(f"task_stop:{request_no}", "stop")
    r.set(f"task_status:{request_no}", "stop")
    return {"message": "Stop signal sent"}

• 优点：支持多实例部署，实时响应终止命令，性能优异；
• 缺点：需要额外部署Redis服务。

方案3：Celery + 消息队列（工业级复杂场景） #

如果你的服务需要任务持久化、重试机制、大规模分布式调度，Celery+Redis/RabbitMQ是最优解：

# celery_tasks.py
from celery import Celery
from celery.schedules import crontab
import requests

app = Celery('scheduled_tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

@app.task(bind=True)
def send_post(self, request_no: str, url: str):
    # 检查任务是否被标记终止
    if app.backend.get(f"task_status:{request_no}") == b"stop":
        self.revoke(terminate=True)
        return
    requests.post(url)

# FastAPI接口
from fastapi import FastAPI

app_api = FastAPI()

@app_api.post("/start-task")
async def start_task(interval: int, request_no: str, total_duration: int, url: str):
    # 动态添加定时任务
    app.add_periodic_task(
        interval,
        send_post.s(request_no, url),
        name=f"task_{request_no}",
        expires=total_duration
    )
    app.backend.set(f"task_status:{request_no}", "running")
    return {"message": "Scheduled task added"}

@app_api.post("/stop-task")
async def stop_task(request_no: str):
    app.backend.set(f"task_status:{request_no}", "stop")
    app.control.revoke(f"task_{request_no}", terminate=True)
    return {"message": "Task revoked"}

• 优点：成熟的分布式任务框架，支持重试、持久化、多实例调度；
• 缺点：学习曲线略高，需要部署Celery Beat和消息中间件。

选型建议 #

• 单实例、需求简单：选方案1，快速落地无额外依赖；
• 多实例、需要实时终止：选方案2，轻量易维护；
• 复杂生产场景（重试、持久化、大规模调度）：选方案3，工业级解决方案。

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