基于Django+Celery+Redis，如何估算已启动任务的剩余时间？

Celery本身不提供任务进度和剩余时间的原生支持，需要自行实现进度上报和时间计算逻辑：

1. 在任务执行过程中，定期上报当前完成的进度（如处理的项目数、百分比）和已消耗时间
2. 在前端轮询任务状态时，基于已上报的进度数据计算剩余时间

方案一：利用Celery的update_state实现进度跟踪 #

1. 修改Celery任务（task.py） #

给任务添加bind=True参数，允许任务实例调用update_state方法上报进度和时间数据：

from celery import Celery
import time

app = Celery('myapp')

@app.task(bind=True)
def new_celery_task(self, arg_1, arg_2):
    # 假设arg_1是待处理的数据集，总长度作为进度总量
    total_items = len(arg_1)
    start_time = time.time()
    
    # 初始化任务状态
    self.update_state(
        state='PROGRESS',
        meta={
            'current': 0,
            'total': total_items,
            'start_time': start_time,
            'status': '任务启动中...'
        }
    )
    
    # 模拟任务执行，每处理一项更新一次进度
    for i, item in enumerate(arg_1):
        # 替换为实际任务逻辑
        time.sleep(0.1)
        
        current_progress = i + 1
        elapsed_time = time.time() - start_time
        
        # 上报最新进度
        self.update_state(
            state='PROGRESS',
            meta={
                'current': current_progress,
                'total': total_items,
                'elapsed_time': elapsed_time,
                'status': f'已处理 {current_progress}/{total_items} 项'
            }
        )
    
    return {'result': '任务执行完成'}

2. 更新视图中的状态轮询逻辑（views.py） #

在poll_task_status中获取任务的进度元数据，计算剩余时间：

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

@csrf_protect
def initiate_task(request):
    # 启动任务并返回task_id给前端
    task = new_celery_task.delay(['item1', 'item2', ...], 'param2')
    return JsonResponse({'task_id': task.id})

@csrf_protect
def poll_task_status(request, task_id):
    task_result = AsyncResult(task_id)
    
    if task_result.ready():
        try:
            result = task_result.get()
            return JsonResponse({
                'status': 'completed',
                'result': result
            })
        except Exception as e:
            return JsonResponse({'status': 'error', 'message': str(e)}, status=500)
    
    elif task_result.state == 'REVOKED':
        return JsonResponse({'status': 'terminated', 'message': '任务已终止'})
    
    elif task_result.state == 'PROGRESS':
        meta = task_result.info
        current = meta.get('current', 0)
        total = meta.get('total', 1)
        elapsed_time = meta.get('elapsed_time', 0)
        
        remaining_time = None
        if current > 0:
            # 计算单步平均耗时，再乘以剩余步数得到剩余时间
            avg_time_per_item = elapsed_time / current
            remaining_items = total - current
            remaining_time = round(avg_time_per_item * remaining_items, 2)
        
        return JsonResponse({
            'status': 'pending',
            'current_progress': current,
            'total_progress': total,
            'est_remaining_time': remaining_time,
            'status_message': meta.get('status', '处理中...')
        })
    
    else:
        # 任务仍在队列等待执行
        return JsonResponse({
            'status': 'pending',
            'est_remaining_time': None,
            'status_message': '等待执行中...'
        })

方案二：用Redis独立存储进度数据 #

如果任务数量大或需要存储更多元数据，可以直接用Redis存储进度信息，避免占用Celery的状态存储：

1. 修改Celery任务（task.py） #

import redis
from celery import Celery
import time

app = Celery('myapp')
# 初始化Redis客户端
redis_client = redis.Redis(host='localhost', port=6379, db=0)

@app.task
def new_celery_task(arg_1, arg_2):
    total_items = len(arg_1)
    start_time = time.time()
    task_id = new_celery_task.request.id
    
    # 初始化任务数据到Redis
    redis_client.hset(f'task_{task_id}', mapping={
        'start_time': start_time,
        'current': 0,
        'total': total_items,
        'status': '任务启动中...'
    })
    
    for i, item in enumerate(arg_1):
        time.sleep(0.1)
        current_progress = i + 1
        elapsed_time = time.time() - start_time
        
        # 更新Redis中的进度数据
        redis_client.hset(f'task_{task_id}', mapping={
            'current': current_progress,
            'elapsed_time': elapsed_time,
            'status': f'已处理 {current_progress}/{total_items} 项'
        })
    
    # 任务完成后清理Redis数据（可选）
    redis_client.delete(f'task_{task_id}')
    return {'result': '任务执行完成'}

2. 更新视图轮询逻辑（views.py） #

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

redis_client = redis.Redis(host='localhost', port=6379, db=0)

@csrf_protect
def initiate_task(request):
    task = new_celery_task.delay(['item1', 'item2', ...], 'param2')
    return JsonResponse({'task_id': task.id})

@csrf_protect
def poll_task_status(request, task_id):
    task_result = AsyncResult(task_id)
    
    if task_result.ready():
        try:
            result = task_result.get()
            return JsonResponse({'status': 'completed', 'result': result})
        except Exception as e:
            return JsonResponse({'status': 'error', 'message': str(e)}, status=500)
    
    elif task_result.state == 'REVOKED':
        return JsonResponse({'status': 'terminated', 'message': '任务已终止'})
    
    else:
        # 从Redis读取任务进度数据
        task_data = redis_client.hgetall(f'task_{task_id}')
        if not task_data:
            return JsonResponse({
                'status': 'pending',
                'est_remaining_time': None,
                'status_message': '等待执行中...'
            })
        
        # 转换Redis返回的bytes类型数据
        current = int(task_data.get(b'current', 0))
        total = int(task_data.get(b'total', 1))
        elapsed_time = float(task_data.get(b'elapsed_time', 0))
        
        remaining_time = None
        if current > 0:
            avg_time_per_item = elapsed_time / current
            remaining_items = total - current
            remaining_time = round(avg_time_per_item * remaining_items, 2)
        
        return JsonResponse({
            'status': 'pending',
            'current_progress': current,
            'total_progress': total,
            'est_remaining_time': remaining_time,
            'status_message': task_data.get(b'status', b'处理中...').decode('utf-8')
        })

优化建议 #

• 不稳定任务的估算优化：如果任务耗时波动大，可以取最近N步的平均耗时（而非总平均）来计算剩余时间，减少误差。
• 队列等待时间估算：如果任务在队列中等待，可以结合Celery队列的任务数量和历史平均执行时间，估算等待时长（需要额外监控队列状态）。
• 进度粒度控制：避免过于频繁上报进度（比如每处理1%的任务才上报一次），减少Redis或Celery状态存储的压力。

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