能否根据任务最大内存占用动态调整Celery worker并发数？

方案1：基于Celery信号的动态内存配额控制 #

该方案无需拆分队列，仅在现有单Worker实例上改造即可实现：

• 为两类任务标注内存上限：在任务装饰器中增加自定义参数，示例如下：

  # A类任务，最高8GB内存
  @app.task(max_memory=8 * 1024 ** 3)
  def task_a():
      pass
  
  # B类任务，最高4GB内存
  @app.task(max_memory=4 * 1024 ** 3)
  def task_b():
      pass
  

• 通过Celery内置信号统计运行中任务的总内存配额：
  • 绑定task_prerun信号：任务启动前，将该任务的max_memory值累加到全局的运行内存总额度变量中
  • 绑定task_postrun信号：任务结束后，从全局运行内存总额度中扣除该任务的max_memory值
• 自定义消费拦截逻辑：在task_received信号中做校验，如果当前运行总内存额度 + 新任务的max_memory超过16GB物理内存上限，就将任务重新推回队列尾部，间隔1~2秒后再尝试拉取新任务
  注意：如果使用prefork模式的Worker，需要用Redis、memcached或者本地共享内存存储全局运行总内存值，做好并发安全控制，避免多进程数据不同步导致额度计算错误

方案2：单机器部署多队列多Worker #

你之前了解的多队列多Worker方案不需要额外服务器，完全可以在当前16GB机器上运行，没有额外硬件成本，且全程使用Celery原生配置，无需自定义开发：

• 为两类任务创建独立队列：A类任务投递到queue_a，B类任务投递到queue_b
• 在同一台机器上启动两个独立的Worker进程：
  • 处理A类任务的Worker：celery -A <你的项目入口> worker -Q queue_a -c 1 --max-memory-per-child=8589934592，并发数设为1，同一时间最多跑1个A类任务，刚好占用8GB内存
  • 处理B类任务的Worker：celery -A <你的项目入口> worker -Q queue_b -c 3 --max-memory-per-child=4294967296，并发数设为3，同一时间最多跑3个B类任务，刚好占用12GB内存
• 两个Worker进程的总内存峰值最高为16GB，完全符合物理内存限制，不会触发交换分区读写

如果你的任务内存峰值稳定，方案2是更稳妥的选择，不需要维护自定义调度逻辑，也不会出现任务重复入队的额外开销。

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