Python Celery：如何无序加入任务结果？答案

【问题标题】：Python Celery: How do I join task results out of order?Python Celery：如何无序加入任务结果？
【发布时间】：2019-02-20 20:16:17
【问题描述】：

我有一个简单的项目，我在其中创建了一堆彼此不相关的工作，创建任务，将它们传递给 Redis，并让许多工作人员分布在 Docker Swarm 上，通过长长的队列咀嚼- 运行任务。当工作人员完成时，他们将完成的工作转储到 NFS 共享中，并将文本值发送回 Celery 客户端。

我在 asyncresult 对象的结果集数组上使用 celery.result.ResultSet 的 .join() 函数。 join() 包含一个回调，（目前）只打印结果。

我的问题是 join() 阻塞，直到它按照给定的顺序接收每个 asyncresult 值。我的 swarm 由许多主机组成，这些主机是完全不同的机器，对我来说重要的是在它们完成时返回结果，而不是按顺序或全部完成后返回。

有没有办法通过 Celery 在任务完成时正确触发回调函数？我在网上查看了很多示例，似乎我唯一的选择就是用 asyncio 试试运气，但 Python 并不是我的强项。

用于创建任务和 ResultSet obj 的函数：

def populateQueue(encodeTasks):
r = ResultSet([])
taskHandles = {}

for task in encodeTasks:
    try:
        ret = encode.delay(task)
        r.add(ret)
        logging.debug("Task ID: " + str(ret.task_id))
        taskHandles[ret.task_id] = ret 
    except:
        logging.info("populateQueue fail: " + str(task.traceback))

logging.info("Tasks queued: " + str(len(taskHandles)))
return taskHandles, r

等待结果的main()部分：

        frameCountTotal = getFrameCount(targetFile)
        encodeTasks = buildCmdString(targetFile, frameCountTotal, clientCount)
        taskHandles, retSet = populateQueue(encodeTasks)

        logging.info("Waiting on tasks...")
        retSet.join(callback=testCallback)

提前致谢

【问题讨论】：

标签： python asynchronous celery

【解决方案1】：

找到了我自己的问题的答案：

ResultSet 有另一种称为 join_native() 的方法，我认为只要该代理是几个已知产品（RabbitMQ、Redis 等）之一，它就会对代理使用更具体的 API 调用。 Celery 的文档只是说，如果您满足经纪人的要求，它会提供更好的性能。文档没有说的是它允许乱序返回（至少在 Redis 上，没有尝试过 RMQ）。

【讨论】：