标签: celerybeat

我有一个部署到Heroku的Django应用程序,其中一个工作进程运行芹菜(+ celerycam用于监控).我使用RedisToGo的Redis数据库作为经纪人.我注意到Redis的内存不足.

这是我的procfile的样子:

web: python app/manage.py run_gunicorn -b "0.0.0.0:$PORT" -w 3
worker: python lipo/manage.py celerycam & python app/manage.py celeryd -E -B --loglevel=INFO

这是KEYS'*'的输出:

1. "_kombu.binding.celeryd.pidbox"
2. "celeryev.643a99be-74e8-44e1-8c67-fdd9891a5326"
3. "celeryev.f7a1d511-448b-42ad-9e51-52baee60e977"
4. "_kombu.binding.celeryev"
5. "celeryev.d4bd2c8d-57ea-4058-8597-e48f874698ca"
6. `_kombu.binding.celery"

celeryev.643a99be-74e8-44e1-8c67-fdd9891a5326 正在填写这些消息:

{"sw_sys": "Linux", "clock": 1, "timestamp": 1325914922.206671, "hostname": "064d9ffe-94a3-4a4e-b0c2-be9a85880c74", "type": "worker-online", "sw_ident": "celeryd", "sw_ver": "2.4.5"}

知道如何定期清除这些消息吗？

我正在使用Celery 4.0.1,Django 1.10我有麻烦调度任务(运行任务工作正常).这是芹菜配置:

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myapp.settings')
app = Celery('myapp')

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

app.conf.BROKER_URL = 'amqp://{}:{}@{}'.format(settings.AMQP_USER, settings.AMQP_PASSWORD, settings.AMQP_HOST)
app.conf.CELERY_DEFAULT_EXCHANGE = 'myapp.celery'
app.conf.CELERY_DEFAULT_QUEUE = 'myapp.celery_default'
app.conf.CELERY_TASK_SERIALIZER = 'json'
app.conf.CELERY_ACCEPT_CONTENT = ['json']
app.conf.CELERY_IGNORE_RESULT = True
app.conf.CELERY_DISABLE_RATE_LIMITS = True
app.conf.BROKER_POOL_LIMIT = 2

app.conf.CELERY_QUEUES = (
    Queue('myapp.celery_default'),
    Queue('myapp.queue1'),
    Queue('myapp.queue2'),
    Queue('myapp.queue3'),
)

然后在tasks.py中我有:

@app.task(queue='myapp.queue1')
def my_task(some_id):
    print("Doing something with", some_id)

在views.py中,我想安排此任务:

def my_view(request, id):
    app.add_periodic_task(10, my_task.s(id))

然后我执行命令:

sudo systemctl start rabbitmq.service
celery -A myapp.celery_app beat -l debug
celery worker -A myapp.celery_app

但是这项任务从未安排过.我在日志中看不到任何内容.这项任务正在起作用,因为如果在我看来我做了:

def my_view(request, …

我使用RedisToGo Nano插件在Heroku上使用celerybeat

有一个网络dyno和一个工人dyno

celerybeat工作人员每分钟都要执行一项任务.

问题是:每当我部署新的提交时,dynos重启,我就会收到此错误

2014-02-27T13:19:31.552352+00:00 app[worker.1]: Traceback (most recent call last):
2014-02-27T13:19:31.552352+00:00 app[worker.1]:   File "/app/.heroku/python/lib/python2.7/site-packages/celery/worker/consumer.py", line 389, in start
2014-02-27T13:19:31.552352+00:00 app[worker.1]:     self.reset_connection()
2014-02-27T13:19:31.552352+00:00 app[worker.1]:   File "/app/.heroku/python/lib/python2.7/site-packages/celery/worker/consumer.py", line 727, in reset_connection
2014-02-27T13:19:31.552352+00:00 app[worker.1]:     self.connection = self._open_connection()
2014-02-27T13:19:31.552352+00:00 app[worker.1]:   File "/app/.heroku/python/lib/python2.7/site-packages/celery/worker/consumer.py", line 792, in _open_connection
2014-02-27T13:19:31.552352+00:00 app[worker.1]:     callback=self.maybe_shutdown)
2014-02-27T13:18:23.864287+00:00 app[worker.1]:     self.on_connect()
2014-02-27T13:18:23.864287+00:00 app[worker.1]:   File "/app/.heroku/python/lib/python2.7/site-packages/redis/connection.py", line 263, in on_connect
2014-02-27T13:18:23.864287+00:00 app[worker.1]:     if nativestr(self.read_response()) != 'OK':
2014-02-27T13:18:23.864287+00:00 app[worker.1]:   File "/app/.heroku/python/lib/python2.7/site-packages/redis/connection.py", line 314, in read_response
2014-02-27T13:18:23.864287+00:00 app[worker.1]:     raise response
2014-02-27T13:18:23.864287+00:00 app[worker.1]: ResponseError: max …

我正在使用芹菜建立应用程序,最近我们要求按计划运行某些任务.

我认为celerybeat是完美的,但我有几个问题:

1. 是否可以运行多个celerybeat实例,以便不重复任务？
2. 如何确保celerybeat始终正常运行？

到目前为止,我读到了这个:https: //github.com/celery/celery/issues/251 和 https://github.com/ybrs/single-beat

它似乎应该运行celerybeat的单个实例.

我正在AWS elasticbeanstalk docker容器中运行应用程序,而芹菜工作者也是docker容器(所以它在需要时可以快速扩展).

最好让celerybeat和芹菜工人一起经营supervisord,但似乎这不是正确的方法.

同时拥有celerybeat的单一实例需要手动提供/启动和监控.

我收到以下错误:

File "/Library/Python/2.7/site-packages/Django-1.8.2-py2.7.egg/django/utils/translation/trans_real.py", line 164, in _add_installed_apps_translations
"The translation infrastructure cannot be initialized before the "
django.core.exceptions.AppRegistryNotReady: The translation infrastructure cannot be initialized before the apps registry is ready. Check that you don't make non-lazy gettext calls at import time.

我有一个项目,它不是一个django应用程序,而是一个芹菜应用程序.因此,我还没有创建一个wsgi.py或models.py或通过创建典型的文件django-admin时,一个项目或应用程序已启动.

我只想使用djcelery能够使用djcelery.schedules.DatabaseScheduler此处指定的类似方法创建周期性任务在运行时添加,修改,删除celery.schedules和此处如何动态添加/删除周期性任务到Celery(celerybeat)

这里给出的问题的解决方案(AppRegistryNotReady,使用uWSGI部署时的转换错误)要求我对vassal.ini文件进行更改.我的实现中没有vassal.ini文件.

我将简要介绍一下我的项目 -

proj
  apps.py
  tasks.py
  celeryconfig.py
  runproj.py

• apps.py

    from celery import Celery
    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celeryconfig')
    myapp = Celery('myapp')
    myapp.config_from_object('celeryconfig')
    if __name__ == '__main__' …

我希望能够让我的应用程序的用户启动/停止Celery beat定期的crontab样式任务.现在我用Celery运行

venv/bin/celery worker -A celery_worker.celery --loglevel=info

我用这个简单的例子运行了Celery Beat:

@celery.task
def add(x, y):
    return x + y

在我的配置文件中:

CELERYBEAT_SCHEDULE = {
    'add-every-30-seconds': {
        'task': 'app.email.add',
        'schedule': timedelta(seconds=30),
        'args': (16, 16)
    },
}
CELERY_TIMEZONE = 'UTC'

然后我用Celery击败了Celery

celery -A celery_worker.celery beat -s ~/Documents/cesco-automation/power/celerybeat-schedule

而且效果很好!但我需要更多地控制时间表.

同样在我的app shell上,我能够做到这一点.

>>>add.apply_async([80,800],countdown=30)

>>> from datetime import datetime, timedelta     
>>> tomorrow = datetime.now() + timedelta(days=1)
>>> add.apply_async(args=[10, 10], eta=tomorrow)

这很棒,但我正在开发一个家庭自动化应用程序,所以我还需要停止任务.我该怎么做呢？？

我还发现这个链接提及有关Django的自定义调度类.这正是我需要的.在Celery文档中,它提到了-S标志,但我不知道如何正确地将类添加到我的Flask应用程序中.我怎样才能和Flask一起使用？

我真的需要Celery Beat吗？除了crontab还有其他选择吗？Crontab似乎不够锐利.

我正在尝试在Elastic Beanstalk上为我的Django应用程序设置日常任务.似乎没有一种可接受的方式来设置它,因为芹菜节拍是Django中定期任务的首选解决方案,但对于负载平衡环境并不是很好.

我已经看到一些解决方案,比如使用leader_only = True设置芹菜节拍,只运行一个实例,但这会留下单点故障.我已经看到其他解决方案允许芹菜的许多实例击败并使用锁来确保只有一个任务通过,但除非失败的实例重新启动,否则最终还是不会完全失败？我看到的另一个建议是有一个单独的实例来运行芹菜节拍,但这仍然是一个问题,除非它有一些方法重新启动它失败.

这个问题有没有合适的解决方案？我宁愿不必照看调度程序,因为很容易注意到我的任务直到一段时间才运行.

我打算将一个项目分解成小型微服务.

所有的微服务都是基于cron的.我认为芹菜作为任务分配以及运行周期性任务的机制(celerybeat).

我不希望每个微服务器构建多个芹菜应用程序,因为这将增加多个代理和多个花系统用于监视的开销.

我尝试在多台服务器上使用单个应用程序,但我失败了 我对芹菜的需求是:

1. 我需要为每个微服务提供独立的服务器
2. 属于某些微服务的任务应该只在他们的服务器上执行; 没有在其他服务器之间共享任务
3. 如果微服务停机,我不希望celerybeat用数千个待处理任务阻塞代理,导致其他微服务停止服务.
4. 在微服务之间不需要任何通信.

我尝试分离每个工作人员的队列,这似乎是不可能的我每个服务器尝试过一个工作人员,但我需要每个微服务器上有一个以上的工作人员

我正在尝试使用 celerybeat 每天在特定时间运行任务。但是出于测试目的，我设置了两个任务每两个小时运行一次，这就是我的配置：

CELERYBEAT_SCHEDULE = {
    'daily-google-connect': {
        'task': 'app.engine.schedule_fetcher',
        'schedule': crontab(hour='*/2'),
        'args': (['G'])
    },
    'daily-facebook-connect': {
        'task': 'app.engine.schedule_fetcher',
        'schedule': crontab(hour='*/2'),
        'args': (['F'])
    }
}

这就是我运行芹菜的方式：

celery beat -A app.engine.celery --schedule=/tmp/celerybeat-schedule --pidfile=/tmp/celerybeat.pid -l info

所有内容都使用 docker-compose 在 Docker 容器中运行，因此我确保重新构建应用程序的映像并重新启动容器。

我什至进入正在运行的容器，我在代码中看到了 crontab 设置...但是在我的日志中，我看到任务每分钟都在运行。

我还能做什么来调试这个？

我很感激任何帮助，

谢谢

我一直在广泛搜索，到目前为止我遇到的一切都表明我的配置是正确的......我可以看到我的任务正在我的 celery 工作容器中注册。我没有看到它们在我的节拍容器中注册（但我认为我应该？我应该只看到我所做的任务）但任务没有运行。

芹菜.py

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'itapp.settings')

app = Celery("itapp")
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

if __name__ == '__main__':
    app.start()

@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

初始化文件

from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this …