如何实现Celery任务执行时暂停，及处理用户页面重载场景？

云上老周

2026-6-18

页面重载时Celery转录任务的处理方案

针对你遇到的页面重载中断转录任务的问题，结合Django+Celery(Redis)技术栈，给出具体实现方案和优化建议：

前端实现要点

1. 用`beforeunload`替代`unload`实现警告与暂停

unload事件触发时机太晚，浏览器可能已经开始卸载页面，导致请求无法发送完成。改用beforeunload事件，它会在页面卸载前触发，能稳定弹出警告并发送暂停请求：

// 全局状态变量
let isTranscribing = false;
let currentTaskId = '';
let pollInterval = null;
let isPausedByUnload = false;

// 监听页面卸载前事件
window.addEventListener('beforeunload', function(e) {
    if (isTranscribing) {
        // 同步发送暂停请求，确保浏览器在卸载前处理完
        const xhr = new XMLHttpRequest();
        xhr.open('POST', `/transcribe/pause_task/${currentTaskId}/`, false);
        xhr.setRequestHeader('X-CSRFToken', getCookie('csrftoken'));
        xhr.send();
        
        isPausedByUnload = true;
        // 触发浏览器默认警告（多数浏览器会忽略自定义文本，返回空字符串也能触发）
        e.preventDefault();
        e.returnValue = '';
        return e.returnValue;
    }
});

// 用户取消卸载后恢复任务
window.addEventListener('focus', function() {
    if (isPausedByUnload && isTranscribing) {
        const xhr = new XMLHttpRequest();
        xhr.open('POST', `/transcribe/resume_task/${currentTaskId}/`, true);
        xhr.setRequestHeader('X-CSRFToken', getCookie('csrftoken'));
        xhr.onload = function() {
            if (xhr.status === 200) {
                isPausedByUnload = false;
            }
        };
        xhr.send();
    }
});

// 获取Django CSRF Token辅助函数
function getCookie(name) {
    let cookieValue = null;
    if (document.cookie && document.cookie !== '') {
        const cookies = document.cookie.split(';');
        for (let i = 0; i < cookies.length; i++) {
            const cookie = cookies[i].trim();
            if (cookie.substring(0, name.length + 1) === (name + '=')) {
                cookieValue = decodeURIComponent(cookie.substring(name.length + 1));
                break;
            }
        }
    }
    return cookieValue;
}

2. 调整轮询函数适配暂停状态

原轮询函数需要保留暂停状态下的轮询，这样后端能感知用户未离开，自动恢复任务：

function pollTaskStatus(taskId) {
    currentTaskId = taskId;
    isTranscribing = true;
    console.log(currentTaskId);
    
    pollInterval = setInterval(() => {
        const xhr = new XMLHttpRequest();
        xhr.onload = function() {
            if (xhr.status == 200) {
                const response = JSON.parse(xhr.responseText);
                switch(response.status) {
                    case 'completed':
                        console.log('转录完成');
                        if (response.reload) window.location.reload();
                        clearInterval(pollInterval);
                        isTranscribing = false;
                        break;
                    case 'paused':
                        // 任务暂停，继续轮询等待恢复
                        break;
                    case 'in_progress':
                        // 正常执行，无操作
                        break;
                    case 'failed':
                        showError('任务执行失败');
                        clearInterval(pollInterval);
                        isTranscribing = false;
                        break;
                }
            } else {
                showError('请求出错，请稍后重试');
                clearInterval(pollInterval);
                isTranscribing = false;
            }
        };
        xhr.onerror = function() {
            showError('网络连接异常，请检查网络后重试');
            clearInterval(pollInterval);
            isTranscribing = false;
        };
        xhr.open('GET', `/transcribe/poll_task_status/${taskId}/`, true);
        xhr.send();
    }, 5000);
}

后端实现要点

1. 扩展数据库模型跟踪状态

给TranscribedDocument模型添加任务ID、最后轮询时间字段，方便状态管理：

# models.py
from django.db import models
from django.utils import timezone

class TranscribedDocument(models.Model):
    STATE_CHOICES = [
        ('not_started', '未开始'),
        ('in_progress', '执行中'),
        ('paused', '已暂停'),
        ('completed', '已完成'),
        ('failed', '失败'),
    ]
    state = models.CharField(max_length=20, choices=STATE_CHOICES, default='not_started')
    session_id = models.CharField(max_length=100, unique=True)
    task_id = models.CharField(max_length=100, blank=True, null=True)  # 存储Celery任务ID
    last_poll_time = models.DateTimeField(default=timezone.now)  # 最后轮询时间
    # 其他原有字段...

2. 新增暂停/恢复任务API

# views.py
from django.http import JsonResponse
from django.utils import timezone
from celery.result import AsyncResult
from django_redis import get_redis_connection
from .models import TranscribedDocument
from .tasks import transcribe_file_task

# 暂停任务
def pause_task(request, task_id):
    if request.method != 'POST':
        return JsonResponse({'status': 'error', 'msg': '仅支持POST请求'}, status=400)
    
    try:
        doc = TranscribedDocument.objects.get(task_id=task_id)
        if doc.state != 'in_progress':
            return JsonResponse({'status': 'error', 'msg': '任务当前状态不支持暂停'})
        
        doc.state = 'paused'
        doc.save()
        # 用Redis存储暂停标记
        r = get_redis_connection('default')
        r.set(f'task_paused:{task_id}', '1')
        return JsonResponse({'status': 'success'})
    except TranscribedDocument.DoesNotExist:
        return JsonResponse({'status': 'error', 'msg': '任务不存在'}, status=404)

# 恢复任务
def resume_task(request, task_id):
    if request.method != 'POST':
        return JsonResponse({'status': 'error', 'msg': '仅支持POST请求'}, status=400)
    
    try:
        doc = TranscribedDocument.objects.get(task_id=task_id)
        if doc.state != 'paused':
            return JsonResponse({'status': 'error', 'msg': '任务当前状态不支持恢复'})
        
        doc.state = 'in_progress'
        doc.save()
        # 清除Redis暂停标记
        r = get_redis_connection('default')
        r.delete(f'task_paused:{task_id}')
        return JsonResponse({'status': 'success'})
    except TranscribedDocument.DoesNotExist:
        return JsonResponse({'status': 'error', 'msg': '任务不存在'}, status=404)

# 修改轮询视图，更新最后轮询时间
def poll_task_status(request, task_id):
    try:
        doc = TranscribedDocument.objects.get(task_id=task_id)
        # 更新最后轮询时间，标记用户仍在页面
        doc.last_poll_time = timezone.now()
        doc.save()
        
        task_result = AsyncResult(task_id)
        if task_result.state == 'SUCCESS':
            doc.state = 'completed'
            doc.save()
            return JsonResponse({'status': 'completed', 'reload': True})
        elif task_result.state == 'FAILURE':
            doc.state = 'failed'
            doc.save()
            return JsonResponse({'status': 'failed'})
        else:
            # 返回数据库中存储的状态（可能是暂停/执行中）
            return JsonResponse({'status': doc.state})
    except TranscribedDocument.DoesNotExist:
        return JsonResponse({'status': 'error', 'msg': '任务不存在'}, status=404)

# 修改提交视图，存储Celery任务ID
def transcribe_submit(request):
    if request.method == 'POST':
        file_name = request.POST.get('file_name')
        file_path = request.POST.get('file_path')
        session_id = request.session.session_key
        
        if not file_name or not file_path:
            return redirect('transcribeSubmit')
        
        try:
            doc = TranscribedDocument.objects.get(session_id=session_id)
            if doc.state == 'not_started':
                doc.state = 'in_progress'
                audio_lang = request.POST.get('audio_language')
                output_type = request.POST.get('output_file_type')
                task = transcribe_file_task.delay(file_path, audio_lang, output_type, 'ai_transcribe_output', session_id)
                doc.task_id = task.id
                doc.save()
            return JsonResponse({'task_id': task.id})
        except TranscribedDocument.DoesNotExist:
            # 处理文档不存在的逻辑
            return redirect('transcribeSubmit')
    # GET请求处理逻辑...

3. 修改Celery任务支持暂停/自动终止

任务内部定期检查暂停标记和用户轮询时间，实现软暂停和自动终止：

# tasks.py
from celery import shared_task
from django.utils import timezone
from django_redis import get_redis_connection
import time
from .models import TranscribedDocument
import os
from django.conf import settings

@shared_task(bind=True)
def transcribe_file_task(self, path, audio_language, output_file_type, dest_dir, session_id):
    r = get_redis_connection('default')
    doc = TranscribedDocument.objects.get(session_id=session_id)
    task_id = self.request.id
    
    try:
        # 初始化输出路径
        local_output_dir = '/tmp/'
        filename = os.path.splitext(os.path.basename(path))[0]
        local_output_file = os.path.join(local_output_dir, f"{filename}.{output_file_type}")

        # 模拟分块处理（实际转录逻辑需要拆分为可中断的步骤）
        if audio_language == 'detect':
            model = load_model()
            audio = load_audio(path)
            audio_chunks = split_audio_into_chunks(audio)  # 自定义拆分音频的函数
            
            for chunk in audio_chunks:
                # 检查是否被暂停
                while r.get(f'task_paused:{task_id}'):
                    time.sleep(1)
                    # 检查用户是否超过15秒未轮询（3次轮询间隔），若超过则终止任务
                    doc.refresh_from_db()
                    time_diff = timezone.now() - doc.last_poll_time
                    if time_diff.total_seconds() > 15:
                        doc.state = 'failed'
                        doc.save()
                        r.delete(f'task_paused:{task_id}')
                        return
                
                # 处理当前音频块
                chunk_result = transcribe(chunk, verbose=False)
                # 可选：保存临时结果
            
            # 合并所有块结果并生成文件
            final_result = combine_chunk_results(chunk_results)  # 自定义合并函数
            writer = get_writer(output_file_type, local_output_dir)
            writer(final_result, path)
            
            # 存储转录结果的逻辑...
            doc.state = 'completed'
            doc.save()
            
    except Exception as e:
        doc.state = 'failed'
        doc.save()
        # 可选：重试任务
        raise self.retry(exc=e, countdown=5)

更优方案建议

替换轮询为WebSocket：用Django Channels实现WebSocket通信，后端可以主动推送任务状态，用户断开连接时能立即感知，无需依赖轮询超时判断，效率更高，状态更实时。
Celery任务软中断：避免直接用revoke强制终止任务，通过Redis标记实现软暂停/终止，保证任务能完成当前步骤，避免数据损坏。
任务状态持久化：所有状态都存在数据库中，不依赖Celery的结果存储，即使Celery重启，任务状态也不会丢失。

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

如何实现Celery任务执行时暂停，及处理用户页面重载场景？

前端实现要点 #

1. 用beforeunload替代unload实现警告与暂停 #

2. 调整轮询函数适配暂停状态 #

后端实现要点 #

1. 扩展数据库模型跟踪状态 #

2. 新增暂停/恢复任务API #

3. 修改Celery任务支持暂停/自动终止 #

更优方案建议 #