Celery结合Amazon SQS FIFO同消息组任务串行执行异常排查

我当前使用Celery作为任务队列，搭配Amazon SQS FIFO服务，期望实现同一消息组ID的任务串行处理，不同消息组ID的任务并行处理。但尽管遵循了推荐配置且了解SQS消息组的特性，同一消息组的任务仍被多个Celery worker进程并行处理。

补充细节 #

• Celery未设置--concurrency参数，默认生成4个池进程（与核心数一致）；
• 通过以下代码传递消息组ID：

message_properties = {
    "MessageGroupId": f"{supplier_id}"
}
celery_task.s(
    param1, **message_properties
).apply_async(**message_properties)

• 已确认队列为FIFO类型（名称以.fifo结尾），额外配置如下：

{
'polling_interval': 60,
'wait_time_seconds': 10,
'visibility_timeout': 600
}

我尝试调整配置但无效，同一消息组ID的任务仍被Celery worker的多个进程并行处理，请问如何实现同一消息组任务串行、不同消息组任务并行的需求？

1. 修正MessageGroupId的传递方式 #

你当前的代码存在参数传递错误：celery_task.s()不需要接收MessageGroupId，这个SQS专属属性必须通过queue_options字典传递给apply_async()，而非当作任务参数传入。正确写法如下：

message_properties = {
    "queue_options": {
        "MessageGroupId": f"{supplier_id}"
    }
}
celery_task.s(param1).apply_async(**message_properties)

之前的写法会让MessageGroupId被当作任务的业务参数，SQS根本识别不到消息组配置，自然无法触发串行逻辑。

2. 限制Celery worker的预取数量 #

Celery默认会预取多个任务到本地进程队列，就算SQS限制了同一消息组的串行分发，已经预取到本地的任务还是会被并行执行。必须调整预取配置：

• 启动worker时添加参数：celery worker --prefetch-multiplier=1
• 或者在Celery配置文件中设置：

CELERYD_PREFETCH_MULTIPLIER = 1

这个配置让每个worker进程每次只预取1个任务，确保同一消息组的下一个任务，只有在上一个任务执行完成并释放SQS可见性超时后，才会被拉取处理。

3. 验证Visibility Timeout的合理性 #

确保visibility_timeout的值大于单个任务的最长执行时间。如果任务执行时间超过这个值，SQS会把未完成的任务重新放回队列，导致同一消息组的任务被多个worker重复拉取并行处理。你当前设置的600秒（10分钟）如果能覆盖任务最长耗时就没问题，否则需要调大该值。

4. 确认队列FIFO特性完整 #

除了名称以.fifo结尾，还要确保队列创建时已开启必要的FIFO配置（比如内容去重或手动传递MessageDeduplicationId），不过这一点不影响消息组的串行逻辑，仅用于避免重复任务。

完成以上调整后，SQS会按消息组顺序分发任务，Celery worker也不会提前预取同一组的多个任务，即可实现同一消息组串行、不同消息组并行的需求。

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