如何不重复拷贝代码实现多队列Celery worker单并发顺序消费

你完全不需要复制多份任务脚本，可按以下方案实现：

方案1：启动命令指定队列（最简便） #

单份Celery任务脚本保持不变，启动worker时通过命令参数区分对接的队列，所有worker复用同一份代码：

• 启动参数说明：
  • -Q参数指定当前worker消费的队列
  • -c 1 配置单并发保证队列内消息顺序消费
  • -n 给不同worker指定唯一的实例名避免冲突
• 启动示例（以3个队列为例）：

  celery -A your_task_module worker -Q queue_1 -c 1 -n worker1@%h
  celery -A your_task_module worker -Q queue_2 -c 1 -n worker2@%h
  celery -A your_task_module worker -Q queue_3 -c 1 -n worker3@%h
  

  其中your_task_module是你定义Celery任务的模块名，所有启动命令共用同一个模块。

方案2：批量启动脚本 #

如果需要启动的worker数量较多，可写一个简单的批量启动脚本，无需手动执行多条命令：

Shell脚本示例 #

#!/bin/bash
# 按需修改worker数量和模块名
WORKER_COUNT=5
TASK_MODULE="your_task_module"

for i in $(seq 1 $WORKER_COUNT)
do
  celery -A ${TASK_MODULE} worker -Q queue_${i} -c 1 -n worker${i}@%h --detach
done

执行该脚本即可一次性启动所有worker实例。

方案3：Python代码内动态启动worker #

如果你需要在Python应用内部直接管理worker生命周期，可通过Celery内置的Worker类配合多进程实现，单份代码即可启动多个worker：

from celery import Celery
from celery.bin.worker import worker as celery_worker
from multiprocessing import Process

# 初始化Celery实例，全局唯一
app = Celery('task_app', broker='你的broker地址')

# 任务定义仅写一次，所有worker复用
@app.task
def your_business_task(params):
    # 你的业务处理逻辑
    pass

# 单个worker启动函数
def run_worker(queue_name: str):
    worker = celery_worker(app=app)
    worker.run(
        queues=[queue_name],
        concurrency=1,
        hostname=f"{queue_name}_worker@%h"
    )

if __name__ == "__main__":
    worker_num = 5
    process_list = []
    # 批量启动多进程worker
    for idx in range(worker_num):
        target_queue = f"queue_{idx + 1}"
        p = Process(target=run_worker, args=(target_queue,))
        p.start()
        process_list.append(p)
    # 等待所有worker进程运行
    for p in process_list:
        p.join()

注意事项 #

• 消息生产侧需要按业务路由规则将消息投递到对应队列，比如将同一业务维度（如同一用户、同一订单）的消息发到同一个队列，即可保证该维度的消息顺序处理。
• 所有方案均无需修改任务逻辑代码，不存在冗余重复代码。

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