如何在用户卸载网页时终止Celery翻译任务？

针对Django+Celery(Redis)架构下，用户页面卸载时终止翻译任务的需求，结合前端可靠请求发送和后端兜底机制实现：

一、前端：确保终止请求能发送成功 #

利用beforeunload事件触发终止请求，同时使用fetch的keepalive属性保证请求在页面卸载后仍能完成，同步清理轮询定时器。

修改原JS代码如下：

let currentTaskId = null;
let pollInterval = null;
let isTranslating = false;

function pollTaskStatus(taskId) {
    currentTaskId = taskId;
    isTranslating = true;
    console.log(currentTaskId);

    // 绑定页面卸载事件
    window.addEventListener('beforeunload', handlePageUnload);

    pollInterval = setInterval(() => {
        const xhr = new XMLHttpRequest();
        xhr.onload = function() {
            if (xhr.status == 200) {
                const response = JSON.parse(xhr.responseText);
                if (response.status === 'completed') {
                    console.log('sent');
                    showTranslationComplete(response);
                    cleanUpTask();
                }
            } else {
                showError('An error occurred.');
                cleanUpTask();
            }
        };
        xhr.onerror = function() {
            showError('Connection error. Please check your network connection and try again.');
            cleanUpTask();
        };
        xhr.open('GET', `/translate/poll_task_status/${taskId}/`, true);
        xhr.send();
    }, 5000); // Poll every 5 seconds
}

function cleanUpTask() {
    clearInterval(pollInterval);
    isTranslating = false;
    window.removeEventListener('beforeunload', handlePageUnload);
    currentTaskId = null;
}

function handlePageUnload() {
    if (isTranslating && currentTaskId) {
        // 发送终止任务请求，keepalive确保请求能完成
        fetch(`/translate/terminate_task/${currentTaskId}/`, {
            method: 'POST',
            keepalive: true,
            headers: {
                'X-CSRFToken': getCsrfToken()
            }
        });
        cleanUpTask();
    }
}

// 辅助函数：获取页面CSRF Token
function getCsrfToken() {
    return document.querySelector('[name=csrfmiddlewaretoken]').value;
}

二、后端：实现终止任务的Django接口 #

在Django中添加视图处理终止请求，调用Celery的revoke方法终止任务：

1. Django视图代码

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from celery import current_app

@csrf_exempt
def terminate_task(request, task_id):
    if request.method == 'POST':
        try:
            # 撤销Celery任务，terminate=True强制终止进程
            current_app.control.revoke(task_id, terminate=True)
            # 可选：在Redis中标记任务已终止，供任务内检查使用
            # redis_client.set(f"task_terminated:{task_id}", "1", ex=3600)
            return JsonResponse({'status': 'success'})
        except Exception as e:
            return JsonResponse({'status': 'error', 'message': str(e)}, status=500)
    return JsonResponse({'status': 'error', 'message': 'Invalid method'}, status=405)

2. URL配置
   在urls.py中添加对应路由：

from django.urls import path
from . import views

urlpatterns = [
    # 其他路由...
    path('translate/terminate_task/<str:task_id>/', views.terminate_task, name='terminate_task'),
]

3. Celery配置注意
   确保Celery配置中开启任务追踪：

# celery.py
CELERY_TASK_TRACK_STARTED = True

三、后端兜底：任务超时与主动检查 #

为避免前端请求失败导致任务无意义运行，添加两种兜底机制：

1. 设置任务超时时间 #

定义Celery任务时，设置软/硬超时时间：

from celery import shared_task
import time

@shared_task(soft_time_limit=300, time_limit=360)  # 软超时5分钟，硬超时6分钟
def translate_file_task(task_id, file_path):
    # 翻译逻辑
    for chunk in process_file(file_path):
        translate_chunk(chunk)
        time.sleep(1)
    return "Translation completed"

2. 任务内主动检查终止标记 #

如果任务运行时间较长，可在执行过程中定期检查Redis中的终止标记，提前退出：

from celery import shared_task
import redis
import time

redis_client = redis.Redis()

@shared_task(soft_time_limit=300, time_limit=360)
def translate_file_task(task_id, file_path):
    for chunk in process_file(file_path):
        # 检查是否被用户终止
        if redis_client.get(f"task_terminated:{task_id}"):
            return "Task terminated by user"
        translate_chunk(chunk)
        time.sleep(1)
    return "Translation completed"

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