Flask 中如何返回 pmt_link 并后台运行 order_status

2026-10-24支付老李

结论

不要在 Flask 请求函数里直接调用持续循环的 order_status(data)。Flask 会等这个函数执行结束后才运行 return result，因此浏览器迟迟收不到支付链接。

更可靠的处理方式是：

创建 pmt_link；
把支付状态跟踪任务交给 Celery、RQ 等后台任务队列；
立即返回 jsonify(link)；
由后台任务轮询支付状态，或通过支付平台的 Webhook 更新订单状态。

生产环境建议使用“任务队列 + Webhook”。线程更适合简单测试或风险较低的场景。

为什么 `asyncio` 没有解决问题

asyncio 可以让适合异步执行的操作在事件循环中协作运行，但不会自动把普通的阻塞循环变成后台任务。

如果 order_status(data) 是同步函数，内部还有类似下面的无限循环：

while True:
    check_payment_status()

直接调用它仍会阻塞当前请求。即使把它改成 async def，只要循环内部使用同步网络请求、time.sleep()，或者请求仍在等待这个协程完成，就无法及时返回响应。

在 Flask 请求中临时使用 asyncio.create_task()，也不适合运行长期任务。请求结束后，任务依赖的事件循环可能停止，无法保证任务继续执行。

推荐方案：使用 Celery 执行后台任务

下面以 Celery 和 Redis 为例。Redis 充当消息代理，Celery worker 则独立于 Flask Web 进程运行。

定义 Celery 应用

# celery_app.py
from celery import Celery

celery = Celery(
    "payment_tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

定义支付状态任务

# tasks.py
from celery_app import celery
from functions import order_status

@celery.task(
    autoretry_for=(Exception,),
    retry_backoff=True,
    max_retries=5,
)
def track_order_status(data):
    return order_status(data)

在 Flask 路由中提交任务

from flask import request, jsonify
from functions import pmt_link
from tasks import track_order_status

@app.route("/addToBasket", methods=["POST"])
def basket_data():
    data = request.get_json(silent=True)

    if not isinstance(data, dict):
        return jsonify({"error": "Invalid JSON body"}), 400

    if data.get("action") != "paymentLink":
        return jsonify({"error": "Unsupported action"}), 400

    link = pmt_link(data)

    # 只提交任务，不等待 order_status 执行完成
    track_order_status.delay(data)

    return jsonify(link), 200

除了启动 Flask 服务，还需要单独启动 Celery worker：

celery -A celery_app.celery worker --loglevel=INFO

具体参数取决于项目目录和模块结构。如果 Celery 找不到 tasks.py，需要显式导入任务模块，或在 Celery 配置中设置任务发现。

更合适的支付状态跟踪方式：Webhook

如果支付服务支持 Webhook，最好让支付平台在订单状态变化时主动通知服务器，而不是让 order_status 一直轮询。

基本流程如下：

浏览器请求支付链接
        ↓
Flask 创建支付订单并立即返回链接
        ↓
用户完成支付
        ↓
支付平台调用 Webhook
        ↓
服务器验证签名并更新订单状态

示例结构：

from flask import request, jsonify

@app.route("/payment/webhook", methods=["POST"])
def payment_webhook():
    payload = request.get_json(silent=True)

    # 必须根据支付平台要求验证签名
    # verify_webhook_signature(request)

    # 根据 payload 中的订单号更新数据库
    # update_order_status(payload)

    return jsonify({"received": True}), 200

Webhook 能避免长期循环和重复查询，也不会让 worker 被永久占用。即使已经使用 Webhook，也可以保留一个有时间限制的后台轮询任务作为补偿机制。

简单场景下使用线程

如果只是本地测试，可以用线程让函数在当前 Flask 进程中运行：

from threading import Thread
from flask import request, jsonify
from functions import pmt_link, order_status

def track_order(data):
    try:
        order_status(data)
    except Exception:
        app.logger.exception("Failed to track order status")

@app.route("/addToBasket", methods=["POST"])
def basket_data():
    data = request.get_json(silent=True)

    if not isinstance(data, dict):
        return jsonify({"error": "Invalid JSON body"}), 400

    if data.get("action") != "paymentLink":
        return jsonify({"error": "Unsupported action"}), 400

    link = pmt_link(data)

    Thread(
        target=track_order,
        args=(data,),
        daemon=True,
    ).start()

    return jsonify(link), 200

这种方案有不少限制：

Flask 或 WSGI worker 重启后，线程任务会丢失；
多进程部署时，任务状态难以统一管理；
无限循环会长期占用线程；
重试、超时和任务监控都不方便；
daemon=True 的线程可能在进程退出时被直接终止。

因此，不应把线程用于重要支付流程的生产环境。

`order_status` 不应无限运行

无论使用线程还是任务队列，都应该为轮询设置超时、间隔和退出条件：

import time

def order_status(data):
    max_attempts = 60
    interval_seconds = 5

    for _ in range(max_attempts):
        status = fetch_order_status(data)

        if status in {"paid", "failed", "cancelled", "expired"}:
            save_order_status(data, status)
            return status

        time.sleep(interval_seconds)

    save_order_status(data, "timeout")
    return "timeout"

还要注意这些问题：

不要把 Flask 的 request、g 或数据库 session 对象直接传给后台任务；
任务参数最好只包含订单 ID 等可序列化数据；
建议传递 order_id，再由后台任务从数据库读取订单；
创建支付订单和提交后台任务时，需要处理重复请求和幂等性；
后台任务必须捕获异常，并记录订单号等必要上下文；
支付结果应以经过签名验证的服务端通知或支付平台查询结果为准，不能信任浏览器提交的状态。

更稳妥的任务调用方式是：

track_order_status.delay(data["order_id"])

不要把完整的请求数据原样交给后台任务。这样既能减少进入消息队列的敏感信息，也能避免任务处理已经过期的数据。