基于Ajax轮询缓解Graphene Django重型请求超时问题

刚好我之前在Graphene Django + Apollo的项目里处理过几乎一样的问题，针对这种边缘场景下的长耗时查询，最稳妥成熟的方案就是异步任务处理 + 客户端轮询，下面我给你拆解具体怎么实现，包括你关心的请求跟踪问题：

一、服务器端核心实现（Graphene Django） #

我们的思路是把耗时计算从同步查询里剥离出来，改成异步任务，给客户端返回一个唯一的任务ID，让客户端通过这个ID轮询状态。

1. 定义任务状态模型与GraphQL Schema #

首先需要一个模型来存储任务的状态、结果和所属用户（可选，用于权限控制），然后用Graphene定义对应的查询和突变：

# models.py
from django.db import models
from django.contrib.auth.models import User

class TaskStatus(models.Model):
    TASK_STATUS_CHOICES = (
        ("PENDING", "Pending"),
        ("SUCCESS", "Success"),
        ("FAILED", "Failed"),
    )
    task_id = models.CharField(max_length=64, unique=True)
    status = models.CharField(max_length=16, choices=TASK_STATUS_CHOICES, default="PENDING")
    result = models.TextField(null=True, blank=True)
    error = models.TextField(null=True, blank=True)
    user = models.ForeignKey(User, on_delete=models.CASCADE, null=True, blank=True)  # 关联用户，可选
    created_at = models.DateTimeField(auto_now_add=True)

# schema.py
import uuid
import graphene
from graphene_django import DjangoObjectType
from .models import TaskStatus
from .tasks import heavy_computation_task

class TaskStatusType(DjangoObjectType):
    class Meta:
        model = TaskStatus
        fields = ("task_id", "status", "result", "error")

class TriggerHeavyComputation(graphene.Mutation):
    class Arguments:
        input_data = graphene.String(required=True)  # 你的业务参数

    task_id = graphene.String()

    def mutate(self, info, input_data):
        # 生成唯一任务ID
        task_id = str(uuid.uuid4())
        # 关联当前登录用户（如果需要权限控制）
        user = info.context.user if info.context.user.is_authenticated else None
        # 创建任务状态记录
        TaskStatus.objects.create(task_id=task_id, user=user)
        # 触发Celery异步任务
        heavy_computation_task.delay(task_id, input_data, user_id=user.id if user else None)
        return TriggerHeavyComputation(task_id=task_id)

class Query(graphene.ObjectType):
    task_status = graphene.Field(TaskStatusType, task_id=graphene.String(required=True))

    def resolve_task_status(self, info, task_id):
        try:
            task_status = TaskStatus.objects.get(task_id=task_id)
            # 权限校验：如果关联了用户，确保当前用户是任务所有者
            if task_status.user and task_status.user != info.context.user:
                return None
            return task_status
        except TaskStatus.DoesNotExist:
            return None

schema = graphene.Schema(query=Query, mutation=TriggerHeavyComputation)

2. 异步任务实现（用Celery） #

用Celery来处理耗时计算，任务执行过程中更新状态：

# tasks.py
from celery import shared_task
from .models import TaskStatus
import time

@shared_task
def heavy_computation_task(task_id, input_data, user_id=None):
    try:
        task_status = TaskStatus.objects.get(task_id=task_id)
        # 模拟你的耗时计算逻辑
        time.sleep(30)  # 替换成实际的复杂数据计算
        result = f"Processed result for input: {input_data}"
        # 更新任务状态为成功
        task_status.status = "SUCCESS"
        task_status.result = result
        task_status.save()
    except Exception as e:
        # 捕获异常，更新状态为失败
        task_status.status = "FAILED"
        task_status.error = str(e)
        task_status.save()

3. 怎么跟踪客户端与请求的关联？ #

其实不需要直接跟踪客户端，唯一的task_id就是最好的标识：每个客户端发起任务时会拿到专属的task_id，后续轮询只需要带上这个ID就能获取自己的任务状态。如果需要更安全的控制，就像上面代码里那样，把task_id和当前登录用户绑定，服务器端校验用户权限，确保用户只能查询自己的任务，防止非法访问。

二、客户端轮询实现（Apollo Client） #

在Apollo里实现轮询非常简单，用useQuery的pollInterval参数就能自动定时查询，步骤如下：

1. 定义GraphQL查询与突变 #

# graphql/operations.js
import { gql } from '@apollo/client';

// 触发异步任务的突变
export const TRIGGER_HEAVY_COMPUTATION = gql`
    mutation TriggerHeavyComputation($inputData: String!) {
        triggerHeavyComputation(inputData: $inputData) {
            taskId
        }
    }
`;

# 查询任务状态的查询
export const GET_TASK_STATUS = gql`
    query GetTaskStatus($taskId: String!) {
        taskStatus(taskId: $taskId) {
            taskId
            status
            result
            error
        }
    }
`;

2. 组件里实现轮询逻辑 #

以React为例，用Apollo的hooks来处理：

import { useState } from 'react';
import { useMutation, useQuery } from '@apollo/client';
import { TRIGGER_HEAVY_COMPUTATION, GET_TASK_STATUS } from './graphql/operations';

function HeavyComputationComponent() {
    const [taskId, setTaskId] = useState(null);
    const [triggerTask, { loading: triggerLoading }] = useMutation(TRIGGER_HEAVY_COMPUTATION);

    // 轮询任务状态，每隔20秒查询一次
    const { data: statusData, loading: statusLoading } = useQuery(GET_TASK_STATUS, {
        variables: { taskId },
        skip: !taskId,  // 没有taskId时不执行查询
        pollInterval: 20000,  // 20秒轮询一次
    });

    const handleStartTask = async () => {
        try {
            const result = await triggerTask({
                variables: { inputData: "your_business_input_here" }
            });
            setTaskId(result.data.triggerHeavyComputation.taskId);
        } catch (err) {
            console.error("Failed to start task:", err);
        }
    };

    if (!taskId) {
        return (
            <button onClick={handleStartTask} disabled={triggerLoading}>
                {triggerLoading ? "Starting..." : "Start Heavy Computation"}
            </button>
        );
    }

    if (statusLoading) {
        return <div>Checking status...</div>;
    }

    const taskStatus = statusData?.taskStatus;
    if (!taskStatus) {
        return <div>Task not found</div>;
    }

    switch (taskStatus.status) {
        case "PENDING":
            return <div>Processing your request... Please wait</div>;
        case "SUCCESS":
            return <div>✅ Result: {taskStatus.result}</div>;
        case "FAILED":
            return <div>❌ Error: {taskStatus.error}</div>;
        default:
            return <div>Waiting for update...</div>;
    }
}

export default HeavyComputationComponent;

三、可选优化：用GraphQL订阅替代轮询 #

如果觉得轮询不够高效（比如任务完成时间不确定，不想浪费请求），可以用GraphQL订阅实现服务器主动推送状态。不过这个需要搭建WebSocket支持，Graphene Django可以结合Django Channels和graphene-subscriptions来实现，任务完成后服务器主动把状态推送给对应的客户端。但这个方案的复杂度比轮询高，如果你20秒的轮询间隔可以接受，轮询是更简单易维护的选择。

四、额外注意事项 #

• 任务清理：定期清理数据库里的旧任务记录，比如用Celery定时任务删除超过7天的TaskStatus数据，避免数据堆积。
• 超时控制：给Celery任务设置超时时间，防止任务无限挂起。
• 错误重试：如果任务可能失败，可以给Celery任务添加重试机制，比如@shared_task(autoretry_for=(Exception,), retry_backoff=3)。

内容的提问来源于stack exchange，提问作者anthony-dandrea