我刚刚发现了配置选项CELERYD_PREFETCH_MULTIPLIER(docs).默认值为4,但(我相信)我希望预取或尽可能低.我现在把它设置为1,这足够接近我正在寻找的东西,但仍有一些我不明白的东西:
为什么这个预取是一个好主意？除非消息队列和工作人员之间存在大量延迟(在我的情况下,他们当前在同一主机上运行,最糟糕的情况可能最终在同一数据中的不同主机上运行),我真的没有理由看到它.中央).文档仅提到了缺点,但未能解释其优点.
许多人似乎把它设置为0,期望能够以这种方式关闭预取(在我看来是一个合理的假设).但是,0表示无限预取.为什么有人会想要无限制的预取,这不是完全消除了你首先引入任务队列的并发/异步性吗？
为什么不能关闭预取？在大多数情况下,关闭性能可能不是一个好主意,但有技术上的原因是不可能的吗？还是只是没有实施？
有时,此选项已连接到CELERY_ACKS_LATE.例如.Roger Hu写道 «[...]通常[用户]真正想要的是让工人只保留与子进程一样多的任务.但是,如果没有启用延迟确认,这是不可能的[...]»我不明白这两个选项是如何连接的,以及为什么没有另一个选项是不可能的.可以在此处找到关于连接的另一个提及.有人可以解释为什么这两个选项是连接的吗？
似乎我让我的Rabbitmq服务器运行的时间越长,我对未确认消息的麻烦就越多.我很乐意将它们重新排列.实际上似乎有一个amqp命令来执行此操作,但它仅适用于您的连接使用的通道.我制作了一个小的鼠兔脚本,至少尝试一下,但是我要么缺少一些东西,要么就是这样做了(用rabbitmqctl怎么样？)
import pika
credentials = pika.PlainCredentials('***', '***')
parameters = pika.ConnectionParameters(host='localhost',port=5672,\
credentials=credentials, virtual_host='***')
def handle_delivery(body):
"""Called when we receive a message from RabbitMQ"""
print body
def on_connected(connection):
"""Called when we are fully connected to RabbitMQ"""
connection.channel(on_channel_open)
def on_channel_open(new_channel):
"""Called when our channel has opened"""
global channel
channel = new_channel
channel.basic_recover(callback=handle_delivery,requeue=True)
try:
connection = pika.SelectConnection(parameters=parameters,\
on_open_callback=on_connected)
# Loop so we can communicate with RabbitMQ
connection.ioloop.start()
except KeyboardInterrupt:
# Gracefully close the connection
connection.close()
# Loop until we're fully closed, will stop on …我在Celery有一个任务,可以在正常运行的情况下运行10,000秒.但是,我的所有其余任务都应该在不到一秒的时间内完成.如何在不更改短期运行任务的时间限制的情况下为故意长时间运行的任务设置时间限制？
在发布新版本以更新工作人员代码的同时,如何优雅地重新启动芹菜工作者？
编辑: 我打算做的是这样的事情.
我们使用Celery和我们的Django webapp来管理离线任务; 其中一些任务可以运行长达120秒.
每当我们进行任何代码修改时,我们都需要重新启动Celery以重新加载新的Python代码.我们当前的解决方案是将SIGTERM发送到主Celery进程(kill -s 15 `cat /var/run/celeryd.pid`),然后等待它死并重新启动它(python manage.py celeryd --pidfile=/var/run/celeryd.pid [...]).
由于长时间运行的任务,这通常意味着关闭将花费一两分钟,在此期间不会处理任何新任务,从而导致当前站点上的用户显着延迟.我正在寻找一种方法告诉Celery关闭,但随后立即启动一个新的Celery实例来开始运行新任务.
事情并没有工作:
ERROR: Pidfile (/var/run/celeryd.pid) already exists. Seems we're already running? (PID: 13214)立即抱怨并死亡.(这看起来像芹菜本身的一个错误;我让他们知道它.)我使用Celery来执行异步后台任务,Redis作为后端.我对Celery工作人员在以下情况下的行为感兴趣:
我正在使用一个守护进程作为守护进程celeryd.已通过该-Q选项为此工作人员分配了两个队列以供使用:
celeryd -E -Q queue1,queue2
工作人员如何决定从哪里获取下一个要使用的任务？是否随机消耗任何一个任务queue1或queue2？它会优先获取,queue1因为它是传递给参数列表中的第一个-Q吗？
我刚刚开始使用django-celery,我想将celeryd设置为守护进程.但是,这些说明似乎表明它一次只能配置一个站点/项目.芹菜可以处理多个项目,还是只能处理一个项目？而且,如果是这种情况,是否有一种干净的方法来设置celeryd为每个配置自动启动,这需要我为每个配置创建一个单独的init脚本？
我Celery在模块中定义了一个应用程序,现在我想从其中的同一个模块启动worker __main__,即通过运行模块python -m而不是celery从命令行运行.我试过这个:
app = Celery('project', include=['project.tasks'])
# do all kind of project-specific configuration
# that should occur whenever this module is imported
if __name__ == '__main__':
# log stuff about the configuration
app.start(['worker', '-A', 'project.tasks'])
但现在Celery认为我正在运行没有参数的工人:
Usage: worker <command> [options]
Show help screen and exit.
Options:
-A APP, --app=APP app instance to use (e.g. module.attr_name)
[snip]
使用消息是您获得的消息celery --help,就好像它没有获得命令一样.我也试过了
app.worker_main(['-A', 'project.tasks'])
但是抱怨-A不被承认.
那我该怎么做？或者,如何将回调传递给worker以使其记录有关其配置的信息？
我像在Docs-Example中那样用芹菜重试:
@task()
def add(x, y):
try:
...
except Exception, exc:
add.retry(exc=exc, countdown=60) # override the default and
# retry in 1 minute
每次重试此作业时,如何增加重试倒计时 - 例如60秒,2分钟,4分钟等等,直到MaxRetriesExceeded被提升为止？
在celeryd-multi的文档中,我们找到了这个例子:
# Advanced example starting 10 workers in the background:
# * Three of the workers processes the images and video queue
# * Two of the workers processes the data queue with loglevel DEBUG
# * the rest processes the default' queue.
$ celeryd-multi start 10 -l INFO -Q:1-3 images,video -Q:4,5 data
-Q default -L:4,5 DEBUG
(从这里:http://docs.celeryproject.org/en/latest/reference/celery.bin.celeryd_multi.html#examples)
什么是一个实际的例子,说明为什么在一个主机上有多个工作程序处理相同的队列是好的,如上例所示？这不是设置并发性的原因吗？
更具体地说,以下两行(A和B)之间是否存在实际差异？:
A:
$ celeryd-multi start 10 -c 2 -Q data
B:
$ celeryd-multi start 1 -c 20 …