使用RabbitMQ作为Broker和Backend时，如何获取已成功Celery任务ID列表？

嘿，我完全懂你的顾虑——就为了拿个成功任务ID列表额外装个Flower确实有点冗余，而且你也好奇能不能直接从RabbitMQ里拿到这些信息。先给你理清楚核心逻辑，再给你几个实用的方案：

先明确：RabbitMQ本身存不了这个列表 #

RabbitMQ的定位是消息传递+临时存储任务结果，它不会持久化保存所有已成功任务的ID列表。默认情况下，任务完成后结果会在result_expires设置的时长后被自动清理，而且它也没有提供批量查询成功任务ID的API。所以想拿到这个列表，得换个思路。

方案1：用Celery信号自定义追踪（无需额外依赖） #

Celery自带信号机制，你可以监听task_success信号，在任务成功时自动把ID记录到一个持久化存储里（比如Redis、SQLite，甚至本地文件），之后直接查询这个存储就能得到所有成功任务的ID列表。

举个用Redis存储的简单例子：

from celery import Celery
from celery.signals import task_success
import redis

# 初始化Celery实例
app = Celery('tasks', broker='amqp://guest@localhost//', backend='amqp://guest@localhost//')
# 初始化Redis连接
r = redis.Redis(host='localhost', port=6379, db=0)

# 监听任务成功信号，记录任务ID
@task_success.connect(sender=None)
def track_successful_task(sender=None, result=None, **kwargs):
    # sender是任务实例，sender.id就是对应的任务ID
    r.sadd('successful_tasks', sender.id)

# 查询所有成功任务ID的方法
def get_successful_task_ids():
    # 返回所有成功任务ID的集合，可根据需求转成列表
    return list(r.smembers('successful_tasks'))

这个方式完全不需要额外安装多余的包（除非你选的存储工具还没装），轻量又可控，适合不想加额外依赖的场景。

方案2：批量检查任务状态（不推荐，效率低） #

如果你的任务结果还没过期，你可以尝试用app.control.inspect()获取节点上的任务信息，但这个方法只能拿到当前节点的任务数据，没办法直接筛选出成功的。另外，你也可以手动维护提交过的任务ID列表，之后逐个用AsyncResult(task_id).state判断是否成功，但这种方式在任务量较大时效率极低，只适合小批量任务的场景。

关于Flower的补充 #

你提到Flower能实现这个功能，其实是因为Flower会主动监听Celery的事件流，并把任务状态（包括成功的ID）存在自己的存储里（默认内存，也可以配置成Redis/数据库持久化）。如果你的项目本来就需要监控工具，那用Flower确实方便，但如果只是为了拿个ID列表，自定义信号追踪肯定是更轻量的选择。

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