芹菜任务在无限循环中重新处理自身

Question

我遇到了一个奇怪的情况，celery 会重新处理已完成的任务。整体设计如下：

Celery Beat：定期拉取文件，如果拉取文件，它会在数据库中创建一个新条目，并将该文件的处理委托给 1 个工作队列中的另一个 celery 任务（这样一次只处理一个文件）

Celery 任务：处理文件，一旦完成就完成，不重试，不循环。

@app.task(name='periodic_pull_file')
def periodic_pull_file():
    for f in get_files_from_some_dir(...):
        ingested_file = IngestedFile(filename=filename)
        ingested_file.document.save(filename, File(f))
        ingested_file.save()
        process_import(ingested_file.id)
        #deletes the file from the dir source
        os.remove(....somepath)

def process_import(ingested_file_id):
    ingested_file = IngestedFile.objects.get(id=ingested_file_id)
    if 'foo' in ingested_file.filename.lower():
        f = process_foo
    else:
        f = process_real_stuff
    f.apply_async(args=[ingested_file_id], queue='import')

@app.task(name='process_real_stuff')
def process_real_stuff(file_id):
    #dostuff

process_foo 和 process_real_stuff 只是一个循环遍历文件一次的函数，一旦完成它就完成了。我实际上可以跟踪它所在位置的百分比，我注意到的有趣的事情是同一个文件不断被反复处理（请注意，这些文件很大，处理速度很慢，需要几个小时才能处理。现在我开始了想知道它是否只是在队列中创建重复任务。当我有 13 个待导入文件要导入时，我检查了我的 redis 队列：

-bash-4.1$ redis-cli -p 6380 llen import
(integer) 13

啊哈，13 岁，我检查了每个排队任务的内容，看看它是否只是重复 ingested_file_ids 使用：

redis-cli -p 6380 lrange import 0 -1

它们都是具有唯一 ingested_file_id 的独特任务。我忽略了什么吗？有什么理由让它完成一项任务->一遍又一遍地循环同一个任务吗？这只是最近才开始发生，没有代码更改。以前的事情过去是非常活泼和无缝的。我知道这也不是来自“失败”的进程，它以某种方式神奇地重试自身，因为它没有在队列中向下移动。即它一次又一次地以相同的顺序接收相同的任务，因此它永远不会触及它应该处理的其他 13 个文件。

注意，这是我的工人：

python manage.py celery worker -A myapp -l info -c 1 -Q import

Answer 1

使用这个

celery -Q your_queue_name 清除