芹菜:每个工人对task_acks_late的不同设置/向芹菜添加自定义选项 [英] Celery: different settings for task_acks_late per worker / add custom option to celery

问题描述

此问题是django + celery:禁用一名工作人员的预取功能，是否有错误?

我遇到了一个芹菜问题(请参阅我跟进的问题)，为了解决这个问题，我希望有两个芹菜工人，每个工人的-concurrency 1值，但是具有task_acks_late的两个不同设置.

I had a problem with celery (see the question that I follow up) and in order to resolve it I'd like to have two celery workers with -concurrency 1 each but with two different settings of task_acks_late.

我当前的方法正在工作，但我认为它不是很漂亮.我正在执行以下操作:

My current approach is working, but in my opinion not very beautiful. I am doing the following:

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

in settings.py of my django project:

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

这使我可以使用以下命令启动芹菜工人:

This allows me to start the celery workers with following commands:

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

如果我可以做些类似的事情，将会更直观:

What would be more intuitive would be if I could do something like:

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

我发现初始化具有不同值的不同芹菜工人，但没有不知道如何将其嵌入到我的django/芹菜上下文中.我必须在哪些文件中添加向解析器添加参数的代码，以及如何使用自定义参数来修改celery设置的task_acks_late.

I found Initializing Different Celery Workers with Different Values but don't understand how to embed this in my django / celery context. In which files would I have to add the code that's adding an argument to the parser and how could I use the custom param to modify task_acks_late of the celery settings.

更新:感谢@Greenev的回答，我设法为芹菜添加了自定义选项.但是，似乎用这种机制更改配置太迟了"，并且没有考虑到香槟.

Update: Thanks to @Greenev's answer I managed to add custom options to celery. However it seems, that changing the config with this mechanism 'arrives too late' and the chagne is not taken into account.

推荐答案

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

One possible solution here is to provide acks_late=True as an argument of the shared_task decorator, given your code from the prior question:

@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 ，但是您可以按如下所示添加命令行参数.

UPD. I haven't got task_acks_late to be set using this approach, but you could add a command line argument as follows.

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

You've already linked to a solution. I can't see any django specifics here, just put the parser.add_argument code to where you have defined your app, given your code from the prior question, you would have something like this:

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 信号处理程序中访问参数值

Then you could access your argument value in celeryd_init signal handler

@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

这篇关于芹菜:每个工人对task_acks_late的不同设置/向芹菜添加自定义选项的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！