Celery多请求复用同一长任务实现问题及阻塞故障排查

你的核心需求是让多个请求复用同一个正在执行的长时Celery任务，避免重复执行，这在批量查询或资源密集型任务中非常实用。先分析下你现有代码的问题，再给出具体的解决方案。

现有代码的问题 #

1. 过早删除Redis任务ID：第一个请求在调用task.wait()完成后立即删除了Redis中的getInfoTaskId，导致后续请求进来时无法找到已完成的任务，只能重新发起新任务，完全失去了复用的意义。
2. 对wait()和state的误解：
   • task.wait()本身就是阻塞方法，不管是第一个还是后续请求，调用它都会等待任务完成，这是正常行为，并非“第二个请求阻塞”的问题——你需要的就是让后续请求等待同一任务的结果。
   • 用task.state代替wait()时，第一个请求会立即返回任务当前状态（比如PENDING），不会等待任务完成，所以你误以为“第一个请求阻塞”，实际是它没等到结果就返回了。

正确实现方案 #

方案1：阻塞式请求（适合允许请求长时间等待的场景） #

这个方案保留你原有的阻塞逻辑，但修复Redis任务ID的管理逻辑，确保后续请求能复用同一任务或其结果：

@main.route('/get-info', methods=['POST'])
def get_info():
    # 先检查是否有缓存的任务结果
    cached_result = redis.get('getInfoResult')
    if cached_result:
        return f"task result is {cached_result.decode('utf-8')}"

    # 检查Redis中是否存在有效任务ID
    task_id = redis.get('getInfoTaskId')
    task = None
    if task_id:
        task = add_together.AsyncResult(task_id)
        # 如果任务已结束（成功/失败/取消），则清除无效任务ID
        if task.state in ['SUCCESS', 'FAILURE', 'REVOKED']:
            redis.delete('getInfoTaskId')
            task = None

    # 没有有效任务时，发起新任务并存储ID
    if not task:
        task = add_together.delay(23, 42)
        redis.set('getInfoTaskId', task.id, ex=600)  # 设置10分钟过期

    try:
        # 等待任务完成，设置超时避免无限阻塞
        result = task.wait(timeout=600)
        # 缓存结果到Redis，让后续请求直接读取
        redis.set('getInfoResult', str(result), ex=600)
    except Exception as e:
        # 任务超时或失败时，清除无效任务ID
        redis.delete('getInfoTaskId')
        return f"Task failed or timed out: {str(e)}"

    return f"task result is {result}"

方案2：非阻塞式轮询（推荐，更适合生产环境） #

如果你的场景不允许请求长时间阻塞（比如客户端有超时限制），可以改成让客户端轮询任务状态的方式，避免服务器资源被长时间占用：

from flask import jsonify

@main.route('/get-info', methods=['POST'])
def get_info():
    # 检查缓存结果
    cached_result = redis.get('getInfoResult')
    if cached_result:
        return jsonify({
            'status': 'completed',
            'result': cached_result.decode('utf-8')
        })

    # 检查有效任务ID
    task_id = redis.get('getInfoTaskId')
    task = None
    if task_id:
        task = add_together.AsyncResult(task_id)
        if task.state in ['SUCCESS', 'FAILURE', 'REVOKED']:
            redis.delete('getInfoTaskId')
            task = None

    # 发起新任务
    if not task:
        task = add_together.delay(23, 42)
        redis.set('getInfoTaskId', task.id, ex=600)

    # 返回任务ID和当前状态，让客户端轮询
    return jsonify({
        'status': task.state,
        'task_id': task.id,
        'result': task.result if task.state == 'SUCCESS' else None
    })

# 新增查询任务结果的接口
@main.route('/task-result/<task_id>', methods=['GET'])
def task_result(task_id):
    task = add_together.AsyncResult(task_id)
    response = {'status': task.state}
    if task.state == 'SUCCESS':
        response['result'] = task.result
        # 缓存结果并清除任务ID
        redis.set('getInfoResult', str(task.result), ex=600)
        redis.delete('getInfoTaskId')
    elif task.state in ['FAILURE', 'REVOKED']:
        response['error'] = str(task.info)
        redis.delete('getInfoTaskId')
    return jsonify(response)

关键实现要点 #

• 保留任务ID直到任务结束/过期：不要在第一个请求完成后立即删除任务ID，让后续请求能找到正在执行或已完成的任务。
• 校验任务状态有效性：避免Redis中遗留已结束的任务ID，导致后续请求处理无效任务。
• 缓存任务结果：任务完成后将结果存入Redis，后续请求直接读取缓存，减少Celery查询开销。
• 处理异常情况：任务超时或失败时，及时清除无效任务ID，让后续请求可以重新发起任务。

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