如何在Flask应用中使用Dramatiq实现函数后台运行？

你当前的实现方式无法达成目标，核心原因是你将@dramatiq.actor装饰的process_data函数嵌套在了路由函数内部。Dramatiq的worker进程启动时需要扫描并加载所有注册的任务actor，嵌套在路由里的函数只有请求触发时才会在Flask进程中临时生成，worker进程完全感知不到该任务的存在，因此不会执行对应的逻辑。

正确实现步骤 #

• 调整actor定义位置，将所有任务函数放在模块顶层，确保worker启动时可以加载到，示例结构如下：

import dramatiq
from flask import Flask

app = Flask(__name__)

# 任务actor必须定义在模块顶层
@dramatiq.actor
def process_data():
    print('hi i am in process_data')
    # 若任务中需要用到Flask的数据库、配置等资源，需手动推入应用上下文
    # with app.app_context():
    #     执行需要上下文的逻辑

@app.route('/work')
def work():
    print("hi i am in work")    
    process_data.send()
    return "hello"

if __name__ == "__main__":
    app.run()

• 分别启动两个服务：
  1. 第一个终端运行Flask应用：python app.py
  2. 第二个终端启动Dramatiq worker，需指定加载你的应用模块（假设上述代码存放在app.py中）：dramatiq app

完成上述操作后，访问/work路由，即可在worker终端看到process_data的打印输出，任务会在后台独立执行，不会阻塞Flask的请求响应。

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