如何在Flask应用中实现Python任务队列？异步执行异常求助

为什么请求会阻塞？ #

你误解了asyncio.run()的作用——这个函数会同步等待传入的协程完全执行完毕才会返回。所以你的代码里调用asyncio.run(schedule_hello(date, time))后，必须等到await asyncio.sleep(delta)结束、短信发送完成，才会执行return redirect(...)，这就是用户需要等待到指定时间才能跳转的原因。

async/await是用来优化单线程内的IO密集型任务的，不是用来实现后台异步任务的。如果要在请求处理过程中触发一个不阻塞响应的后台任务，需要用线程或者专门的任务队列。

快速修复：用线程实现后台执行 #

如果是开发测试场景，可以用Python的threading模块把异步任务丢到后台线程，让主线程立即返回重定向：

import threading
import asyncio

@bp.route('/calendar', methods=('GET', 'POST'))
def calendar():
    if request.method == 'POST':
        # ... 省略表单验证、数据库插入代码 ...
        
        # 把异步任务放到后台线程执行
        threading.Thread(
            target=lambda: asyncio.run(schedule_hello(date, time)),
            daemon=True  # 守护线程，主进程退出时自动终止
        ).start()
        
        return redirect(url_for('blog.index'))
    
    return render_template('blog/calendar.html')

注意：这种方式适合简单场景，但守护线程会在Flask进程重启/关闭时被终止，可能导致未执行的任务丢失；另外开发服务器（flask run）是单线程的，生产环境建议用多进程服务器（如Gunicorn）。

生产环境推荐：用任务队列实现定时任务 #

对于需要持久化、可靠的定时任务，推荐使用专业的任务队列，比如Celery或RQ（Redis Queue）。

方案1：Celery（功能强大，适合复杂场景） #

1. 安装依赖

pip install celery redis

2. 配置Celery实例
   在你的Flask项目根目录创建celery_config.py：

from celery import Celery
from flask import Flask

def create_app():
    app = Flask(__name__)
    # ... 你的Flask配置 ...
    return app

app = create_app()
celery = Celery(
    app.import_name,
    broker='redis://localhost:6379/0',  # Redis作为消息中间件
    backend='redis://localhost:6379/0'  # 存储任务结果（可选）
)
celery.conf.update(app.config)

3. 定义定时任务
   创建tasks.py：

from datetime import datetime
from celery_config import celery

def send_sms():
    # 这里编写实际发送短信的代码
    print('Message sent')

@celery.task
def schedule_sms(date, time):
    datetime_str = f"{date} {time}"
    target_dt = datetime.strptime(datetime_str, '%m/%d/%y %H:%M:%S')
    delay = (target_dt - datetime.now()).total_seconds()
    
    if delay <= 0:
        # 目标时间已过，立即执行
        send_sms()
    else:
        # 设置延迟执行
        send_sms.apply_async(countdown=delay)

4. 在路由中调用任务

from tasks import schedule_sms

@bp.route('/calendar', methods=('GET', 'POST'))
def calendar():
    if request.method == 'POST':
        # ... 省略表单验证、数据库插入代码 ...
        
        # 触发定时任务
        schedule_sms.delay(date, time)
        
        return redirect(url_for('blog.index'))
    
    return render_template('blog/calendar.html')

5. 启动Celery Worker
   在终端执行：

celery -A celery_config worker --loglevel=info

方案2：RQ（轻量简洁，适合简单场景） #

1. 安装依赖

pip install rq redis

2. 定义任务
   创建tasks.py：

from datetime import datetime
import time

def send_sms():
    # 编写发短信的代码
    print('Message sent')

def schedule_sms(date, time):
    datetime_str = f"{date} {time}"
    target_dt = datetime.strptime(datetime_str, '%m/%d/%y %H:%M:%S')
    delay = (target_dt - datetime.now()).total_seconds()
    
    if delay > 0:
        time.sleep(delay)
    send_sms()

3. 在路由中调用任务

from rq import Queue
from redis import Redis
import tasks

# 初始化Redis连接
redis_conn = Redis(host='localhost', port=6379, db=0)
q = Queue(connection=redis_conn)

@bp.route('/calendar', methods=('GET', 'POST'))
def calendar():
    if request.method == 'POST':
        # ... 省略表单验证、数据库插入代码 ...
        
        # 将任务加入队列
        q.enqueue(tasks.schedule_sms, date, time)
        
        return redirect(url_for('blog.index'))
    
    return render_template('blog/calendar.html')

4. 启动RQ Worker
   在终端执行：

rq worker

注意事项 #

• 任务队列需要依赖中间件（如Redis），确保服务正常运行。
• 生产环境中，Flask应使用多进程/多线程服务器（如Gunicorn、uWSGI），避免单线程阻塞。
• 任务队列的任务是持久化的，即使Flask进程重启，未执行的任务依然会保留在队列中。

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