python - Celery:每个工作人员的 task_acks_late 的不同设置/向 celery 添加自定义选项

Question

此问题是 django + celery: disable prefetch for one worker, Is there a bug? 的后续问题

我在 celery 方面遇到了问题(请参阅我跟进的问题)，为了解决它，我希望有两个 celery 工作人员，每个工作人员的并发数为 1，但有两个不同的 task_acks_late 设置。

我目前的方法是有效的，但在我看来不是很漂亮。我正在执行以下操作:

在我的 django 项目的 settings.py 中:

CELERY_TASK_ACKS_LATE = os.environ.get("LACK", "False") == "True"

这允许我使用以下命令启动 celery 工作人员:

LACK=True celery -A miniclry worker --concurrency=1 -n w2 -Q=fast,slow --prefetch-multiplier 1 
celery -A miniclry worker --concurrency=1 -n w1 -Q=fast

如果我能做类似的事情，会更直观:

celery -A miniclry worker --concurrency=1 -n w2 -Q=fast,slow --prefetch-multiplier 1 --late-ack=True
celery -A miniclry worker --concurrency=1 -n w1 -Q=fast --late-ack=False

我找到了Initializing Different Celery Workers with Different Values但不明白如何将其嵌入到我的 django/celery 上下文中。我必须在哪些文件中添加向解析器添加参数的代码，以及如何使用自定义参数来修改 celery 设置的 task_acks_late 。

更新:感谢@Greenev 的回答，我设法向 celery 添加自定义选项。然而，似乎用这种机制更改配置“来得太晚了”并且没有考虑到 chagne。

Answer 1

这里的一个可能的解决方案是提供 acks_late=True 作为 shared_task 装饰器的参数，给定上一个问题中的代码:

@shared_task(acks_late=True)
def task_fast(delay=0.1):
    logger.warning("fast in")
    time.sleep(delay)
    logger.warning("fast out")

UPD。我尚未使用此方法设置 task_acks_late，但您可以添加命令行参数，如下所示。

您已经链接到解决方案。我在这里看不到任何 django 细节，只需将 parser.add_argument 代码放入您定义 app 的位置，根据上一个问题中的代码，您将拥有像这样的东西:

app = Celery("miniclry", backend="rpc", broker="pyamqp://")
app.config_from_object('django.conf:settings', namespace='CELERY')

def add_worker_arguments(parser):
    parser.add_argument('--late-ack', default=False)

app.user_options['worker'].add(add_worker_arguments)

然后您可以在 celeryd_init 信号处理程序中访问您的参数值

@celeryd_init.connect
def configure_worker(sender=None, conf=None, options=None, **kwargs):
    conf.task_acks_late = options.get('late-ack') # get custom argument value from options