我完全懂你这种被Celery任务爆炸支配的痛苦——Beat周期性任务不管队列死活疯狂塞重复任务，worker忙起来连inspect都卡成狗，任务OOM死了还留着状态烂摊子，确实闹心。既要兼顾框架的便利性，又要避免重复提交和阻塞，这里有几个亲测有效的稳健方案，不管用不用Celery都能解决问题：

方案一：用Celery实现“单任务自调度+分布式锁”，彻底抛弃Beat和Inspect阻塞 #

核心思路是：不要用Beat每隔30秒硬塞任务，而是让任务自己完成后调度下一次执行；同时用分布式锁替代Celery inspect来确保任务唯一性，彻底解决inspect阻塞的问题。

首先需要一个分布式锁（比如用Redis，因为Celery通常配Redis/RabbitMQ做broker，Redis实现锁很方便），用来防止多个检查任务同时运行：

import redis
from celery import Celery
from datetime import timedelta

app = Celery(..., broker='redis://localhost:6379/0')
app.autodiscover_tasks()

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

def acquire_lock(lock_name, timeout=35):
    # 分布式锁：setnx成功返回True（抢到锁），否则False；同时设置过期时间避免死锁
    return redis_client.set(lock_name, 'locked', ex=timeout, nx=True)

def release_lock(lock_name):
    redis_client.delete(lock_name)

@app.task(bind=True, max_retries=3)
def check_and_submit_pdf_tasks(self):
    lock_key = "pdf_checker_lock"
    # 先抢全局检查锁，抢不到说明有另一个检查任务在跑，直接返回
    if not acquire_lock(lock_key):
        return "Another checker task is running, skipping this round"
    
    try:
        # 1. 查询数据库中待处理的PDF（可包含超时未完成的"正在处理"任务）
        unprocessed_pdfs = get_unprocessed_pdfs()  # 你自己的数据库查询逻辑
        
        # 2. 为每个PDF提交任务，用PDF ID做专属锁，避免同一PDF的任务重复入队
        for pdf in unprocessed_pdfs:
            task_lock_key = f"pdf_task_{pdf.id}"
            # 抢PDF专属锁，锁超时设为任务最长预期执行时间（比如5分钟）
            if acquire_lock(task_lock_key, timeout=300):
                # 任务完成后自动释放PDF锁
                create_pages_for_pdf.apply_async(
                    args=[pdf.id], 
                    link=release_pdf_task_lock.s(task_lock_key)
                )
        
        # 3. 调度下一次检查，30秒后执行
        check_and_submit_pdf_tasks.apply_async(countdown=30)
    except Exception as e:
        # 任务失败，10秒后重试一次
        self.retry(exc=e, countdown=10)
    finally:
        # 释放全局检查锁
        release_lock(lock_key)

@app.task
def release_pdf_task_lock(task_lock_key):
    # 任务完成后释放PDF专属锁
    release_lock(task_lock_key)

@app.task(bind=True)
def create_pages_for_pdf(self, pdf_id):
    try:
        # 1. 标记PDF为正在处理（配合数据库超时检查，应对任务被杀的情况）
        mark_pdf_as_processing(pdf_id)
        
        # 2. 你的业务逻辑：生成PDF页面
        # ...
        
        # 3. 标记PDF为处理完成
        mark_pdf_as_done(pdf_id)
    except Exception as e:
        # 任务失败，标记PDF为待处理，60秒后重试最多2次
        mark_pdf_as_pending(pdf_id)
        self.retry(exc=e, max_retries=2, countdown=60)

这个方案的优势：

• 不需要Celery Beat，完全由任务自己调度下一次执行，避免无脑重复提交
• 用分布式锁替代inspect，彻底解决inspect阻塞的问题
• 锁超时机制自动处理任务OOM/被杀的情况，不会永远卡住任务状态
• 逻辑贴合Celery生态，不用完全脱离框架

方案二：独立守护进程+分布式锁，完全可控无框架黑盒 #

如果你觉得Celery的调度逻辑太受限，可以写一个简单的Python守护进程，用循环+分布式锁控制任务提交，逻辑完全透明：

import time
import redis
from celery import Celery

app = Celery(..., broker='redis://localhost:6379/0')
app.autodiscover_tasks()

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

def acquire_lock(lock_name, timeout=3600):
    return redis_client.set(lock_name, 'running', ex=timeout, nx=True)

def release_lock(lock_name):
    redis_client.delete(lock_name)

def pdf_processor_daemon():
    daemon_lock_key = "pdf_processor_daemon_lock"
    # 确保只有一个守护进程在运行
    if not acquire_lock(daemon_lock_key):
        print("Another daemon is already running, exiting")
        return
    
    try:
        while True:
            # 1. 查询待处理PDF（包含超时未完成的任务）
            unprocessed_pdfs = get_unprocessed_or_timeout_pdfs()  # 自定义查询逻辑
            
            for pdf in unprocessed_pdfs:
                task_lock_key = f"pdf_task_{pdf.id}"
                if acquire_lock(task_lock_key, timeout=300):
                    create_pages_for_pdf.apply_async(
                        args=[pdf.id], 
                        link=release_pdf_task_lock.s(task_lock_key)
                    )
            
            # 2. 没有任务就休眠30秒
            time.sleep(30)
    except KeyboardInterrupt:
        print("Daemon stopping...")
    finally:
        release_lock(daemon_lock_key)

if __name__ == "__main__":
    pdf_processor_daemon()

这个方案的优势：

• 没有Celery Beat的黑盒，所有调度逻辑完全可控
• 只在有任务时提交，不会无脑重复调度
• 结合数据库的超时状态检查，完美处理任务OOM/被杀的遗留问题
• 不需要依赖Celery的复杂特性，代码轻量化

方案三：用Celery第三方插件实现任务唯一性（进阶简化） #

如果你不想自己写锁逻辑，可以用celery-once插件，它专门用来确保同一参数的任务不会重复入队：

首先安装插件：pip install celery-once

然后配置使用：

from celery import Celery
from celery_once import QueueOnce

app = Celery(..., broker='redis://localhost:6379/0')
# 配置celery-once的Redis后端
app.conf.ONCE = {
    'backend': 'celery_once.backends.Redis',
    'settings': {
        'url': 'redis://localhost:6379/0',
        'default_timeout': 300  # 锁超时5分钟
    }
}

# 给PDF任务加唯一性限制
@app.task(base=QueueOnce, once={'graceful': True})
def create_pages_for_pdf(pdf_id):
    # 你的业务逻辑
    pass

# 检查任务简化为自调度模式
@app.task
def check_and_submit_pdf_tasks():
    unprocessed_pdfs = get_unprocessed_pdfs()
    for pdf in unprocessed_pdfs:
        create_pages_for_pdf.apply_async(args=[pdf.id])
    # 30秒后调度下一次检查
    check_and_submit_pdf_tasks.apply_async(countdown=30)

celery-once会自动为同一参数的任务生成唯一锁，重复提交时会直接跳过，任务完成或超时后自动释放锁，完全不用自己处理锁的细节。

总结 #

所有方案都避开了Celery inspect的使用，彻底解决了inspect阻塞的问题，同时针对任务爆炸、OOM遗留状态等痛点做了处理：

• 若想继续用Celery生态：优先选方案一（自调度+分布式锁），逻辑清晰无黑盒；或方案三用插件快速实现唯一性。
• 若追求完全可控：选方案二的独立守护进程，代码轻量化且逻辑透明。

核心原则都是：用分布式锁避免重复提交、抛弃无脑周期性调度改用“任务驱动调度”、用超时机制处理任务异常被杀的情况。

内容来源于stack exchange