使用Celery+Redis实现subprocess异步调用失败，Django视图阻塞求助

核心问题出在视图代码中的task.get()方法——这个方法会同步阻塞当前请求线程，强制等待Celery任务完全执行完毕并返回结果，直接抵消了Celery的异步执行特性，导致API请求必须等到subprocess命令完成才会响应。

1. 移除阻塞逻辑，立即返回任务提交状态 #

修改视图代码，删除task.get()，直接返回任务ID，让客户端通过后续接口查询任务结果：

@csrf_exempt
def start_container1(request):
    if request.method == 'POST':
        from Visiotwin_Configuration.tasks import run_docker_command
        # 建议用命令列表替代shell字符串，避免安全风险
        command_args = ["az", "container", "start", "-n", "<container-name>", "-g", "<res-group-name>"]
        try:
            task = run_docker_command.apply_async(args=[command_args])
            return JsonResponse({
                "status": "任务已提交",
                "task_id": task.id
            }, safe=False)
        except Exception as e:
            return JsonResponse({
                "status": "提交任务失败",
                "error": str(e)
            }, safe=False, status=500)

2. 添加任务状态查询接口 #

新增一个视图，用于根据任务ID查询Celery任务的执行状态和结果：

from celery.result import AsyncResult
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt
def get_task_status(request, task_id):
    if request.method == 'GET':
        task_result = AsyncResult(task_id)
        response_data = {
            "task_id": task_id,
            "status": task_result.status,  # 可选状态：PENDING/STARTED/SUCCESS/FAILURE
            "result": task_result.result if task_result.status == 'SUCCESS' else None
        }
        return JsonResponse(response_data)

在urls.py中配置对应路由：

path('task-status/<str:task_id>/', get_task_status, name='get_task_status'),

3. 优化Celery任务实现 #

3.1 消除shell=True的安全风险 #

shell=True存在命令注入风险，改为直接传递命令参数列表：

from celery import shared_task
from celery.utils.log import get_task_logger
import subprocess

logger = get_task_logger(__name__)

@shared_task
def run_docker_command(command_args):
    try:
        process = subprocess.Popen(
            command_args,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True  # 直接返回字符串，无需手动decode
        )
        output, error = process.communicate()
        if process.returncode == 0:
            logger.info(f"命令执行成功: {output}")
            return {"status": "success", "output": output}
        else:
            logger.error(f"命令执行失败: {error}")
            return {"status": "failed", "error": error}
    except OSError as e:
        error_msg = f"执行命令出错: {str(e)}"
        logger.error(error_msg)
        return {"status": "error", "error": error_msg}

3.2 用Celery日志替代print #

使用Celery内置日志系统，方便后续排查任务执行问题，避免print输出丢失。

1. 确保Celery Worker正常启动：celery -A 你的Django项目名 worker --loglevel=info
2. 调用start_container1接口，此时会立即返回任务ID，不会等待命令执行
3. 通过get_task_status接口传入任务ID，查询任务的实时状态和最终结果

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