Celery子任务间歇性Pending无法执行问题排查求助

数据老陈

2026-6-13

问题描述

环境配置

Celery 搭配 Redis + RabbitMQ 作为 Broker，Flower 用于监控

Celery Worker 启动命令：

source /usr/src/app/Configs/environ2.sh && celery -A Application.tasks worker --pool=gevent -E --loglevel=INFO --concurrency=6

业务流程

文档上传API触发第一个Celery任务call_upload处理初始批次，该任务内部的upload函数通过ThreadPoolExecutor触发第二个Celery任务parallel_batch_process，拆分批次并行处理子批次。

异常现象

第二个任务parallel_batch_process间歇性处于Pending状态，始终不执行且无日志输出；偶尔能被Worker正常拾取
Flower显示所有Worker就绪，第一个任务执行正常且能触发第二个任务，Worker日志无错误/警告
已尝试将两个任务放在同一队列，问题仍未解决

相关代码

第一个Celery任务

@CELERY.task(max_retries=0)
def call_upload(q_id, b_id, user_id,uploaded_zip_folders,batch_display_name,batch_display_id,e_id,method,token,kwargs):
    message, return_response,batch_id = o_uploader.upload(q_id, b_id, user_id,uploaded_zip_folders,batch_display_name,batch_display_id,e_id,kwargs)

第二个任务触发逻辑（upload函数内）

try:
    # Iterating through the Jobs assigned to a Batch.
    with ThreadPoolExecutor(max_workers=4) as executor:  # Adjust the number of workers as needed
        futures = []
        for folder_id, job_ids in folder_job_ids.items():
            f = executor.submit(self.parallel_batch_process,job_ids,b_id,q_id,folder_id,kwargs,logger)
            f.add_done_callback(
                lambda fut, logger=logger: (
                    logger.exception("Error inside parallel_batch_process")
                    if fut.exception()
                    else None
                )
            )
            futures.append(f)
        failures = 0
        for future in as_completed(futures):
            try:
                result = future.result()
                if result == -2:
                    failures += 1
                    logger.error(f"UID: {uid} -- A folder job failed during parallel execution")
            except Exception as e:
                failures += 1
                logger.error(f"UID: {uid} -- An error occurred during parallel execution: {str(e)}")

        if failures > 0:
            return -2
            

    return 1

分析与解决方案

可能的根因

Gevent协程与ThreadPoolExecutor的冲突
Worker使用--pool=gevent协程池，而Gevent的异步模型与线程池（ThreadPoolExecutor）存在兼容性问题，可能导致任务提交逻辑阻塞或异常，使得parallel_batch_process无法被正确发送到Broker。
任务提交的线程安全问题
Celery任务提交（delay()/apply_async()）在多线程环境下可能存在线程本地存储（TLS）冲突，导致任务元数据无法正确传递，进而任务无法被Broker正确路由。
Broker间歇性异常
虽然Flower显示Worker就绪，但Redis/RabbitMQ可能存在连接池耗尽、消息路由异常等情况，导致任务存入队列后无法被Worker拾取。

调试步骤

开启Celery Debug级日志
修改Worker启动命令，提升日志级别：
```
source /usr/src/app/Configs/environ2.sh && celery -A Application.tasks worker --pool=gevent -E --loglevel=DEBUG --concurrency=6
```
观察任务提交阶段的日志，确认是否有任务发送到Broker的记录，以及是否存在连接相关错误。
检查Broker队列状态
- RabbitMQ：执行rabbitmqctl list_queues查看目标队列的消息数，确认Pending任务是否真的被存入队列
- Redis：执行redis-cli LLEN <queue_name>查看队列长度，验证任务是否成功提交
替换ThreadPoolExecutor为直接提交任务
暂时移除ThreadPoolExecutor，直接在upload函数中用parallel_batch_process.delay()提交任务，观察是否还会出现Pending情况，排查线程池是否是问题根源。
测试Gevent与线程池的兼容性
编写简单测试脚本，在Gevent环境中用ThreadPoolExecutor提交Celery任务，验证是否存在环境冲突导致的任务提交异常。

最佳实践

统一并发模型
若Worker使用Gevent协程池，任务内部尽量用gevent.pool.Pool替代ThreadPoolExecutor，保持并发模型一致，避免协程与线程的兼容性问题。
确保任务提交的线程安全
若必须使用线程池，显式创建独立的Celery实例用于线程中的任务提交，避免线程本地存储的冲突。

使用Celery原生分组任务
用Celery自带的group实现并行任务调度，比自定义线程池更适配Celery的调度逻辑：

from celery import group

# 替换ThreadPoolExecutor逻辑
job_group = group(
    parallel_batch_process.s(job_ids, b_id, q_id, folder_id, kwargs, logger)
    for folder_id, job_ids in folder_job_ids.items()
)
results = job_group.apply_async()
failures = sum(1 for res in results.get() if res == -2)

监控Broker健康状态
给Redis/RabbitMQ添加监控告警，关注连接数、队列长度、消息丢弃情况，及时发现Broker层面的异常。

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

Celery子任务间歇性Pending无法执行问题排查求助

环境配置 #

业务流程 #

异常现象 #

相关代码 #

第一个Celery任务 #

第二个任务触发逻辑（upload函数内） #

可能的根因 #

调试步骤 #

最佳实践 #