如何在独立Conda环境中分别运行Flask应用与Celery任务？

完全可以实现Flask应用和Celery任务在独立Conda环境运行，核心是让Flask端只做Celery客户端，不依赖任务的实现代码和依赖包。

解决思路 #

问题出在你当前的Flask代码导入了tasks.py，而该文件依赖torch。实际上Celery的客户端（Flask端）和worker端是完全解耦的：

• Worker端（env1）负责执行任务，需要安装所有任务依赖（比如torch），并定义任务逻辑。
• 客户端（Flask/env2）只需要连接broker和backend，通过任务名称调用任务，不需要知道任务的具体实现。

修改后的代码示例 #

1. 保持tasks.py（env1环境）不变 #

import torch
from celery import Celery

celery_app = Celery('tasks',
                    broker='amqp://usr:pw@localhost:5672/',
                    backend='redis://localhost/0')

@celery_app.task
def analyze(filepath):
    # 这里写你的文件处理逻辑，使用torch
    # 示例返回处理后的文件路径
    new_filepath = f"processed_{filepath}"
    return new_filepath

2. 修改app.py（env2环境），不导入tasks.py #

from flask import Flask, jsonify, url_for
from celery import Celery

app = Flask(__name__)

# 仅配置Celery的broker和backend，无需定义任务
celery_app = Celery('tasks',
                    broker='amqp://usr:pw@localhost:5672/',
                    backend='redis://localhost/0')

@app.route('/a/<path>')
def evoke_task(path):
    # 通过任务全名（tasks.analyze）调用任务
    task = celery_app.send_task('tasks.analyze', kwargs={'filepath': path})
    return jsonify({}), 202, {'Location': url_for('taskstatus', task_id=task.id)}

@app.route('/status/<task_id>')
def taskstatus(task_id):
    # 查询任务状态和结果
    task = celery_app.AsyncResult(task_id)
    if task.state == 'PENDING':
        response = {'state': task.state, 'status': '任务等待中...'}
    elif task.state == 'SUCCESS':
        response = {'state': task.state, 'result': task.result}
    else:
        response = {'state': task.state, 'status': str(task.info)}
    return jsonify(response)

启动方式（和之前一致） #

• 在env1环境启动Celery worker：

  (env1) C:\example>celery -A tasks.celery_app worker -l INFO -P threads
  

• 在env2环境启动Flask：

  (env2) C:\example>flask --app app.py run
  

其他可选方案（若不想用Celery） #

如果需要替代方案，以下几种跨进程任务触发方式也可靠：

• Redis队列（RQ）：类似Celery，Flask端往Redis队列添加任务，env1的worker进程处理任务，依赖仅在worker环境安装。
• 直接使用RabbitMQ：Flask用pika库发送消息到RabbitMQ队列，env1写独立消费者进程处理消息，结果存Redis，Flask再查询Redis获取结果。
• HTTP接口调用：在env1运行一个轻量API服务（比如FastAPI），暴露文件处理接口，Flask通过HTTP请求调用该接口，后台处理后返回结果（适合简单场景，解耦性稍弱）。

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