Подтвердить что ты не робот

Определить, доступен ли сельдерей/работает

Я использую Celery для управления асинхронными задачами. Иногда, однако, процесс сельдерея снижается, что не приводит к тому, что ни одна из задач не выполняется. Я хотел бы иметь возможность проверить статус сельдерея и убедиться, что все работает нормально, и если я обнаруживаю какие-либо проблемы, выдает сообщение об ошибке пользователю. Из документации Celery Worker, похоже, я мог бы использовать ping или inspect для этого, но ping чувствует себя взломанным, и неясно, как именно будет использоваться проверка (если проверка(). register() пуста?).

Любое руководство по этому вопросу будет оценено по достоинству. В принципе, я ищу такой метод:

def celery_is_alive():
    from celery.task.control import inspect
    return bool(inspect().registered()) # is this right??

РЕДАКТИРОВАТЬ: он даже не похож на зарегистрированный() доступен на сельдерее 2.3.3 (даже при том, что в нем указаны документы 2.1). Возможно, пинг - правильный ответ.

РЕДАКТИРОВАТЬ: Ping также, похоже, не делает то, что я думал, что это будет делать, поэтому все еще не уверен ответ здесь.

4b9b3361

ОТВЕТЫ

Ответ 1

Вот код, который я использовал. celery.task.control.Inspect.stats() возвращает dict, содержащий множество подробностей о доступных в настоящее время рабочих, None, если нет работающих рабочих, или вызывает IOError, если он не может подключиться к брокеру сообщений. Я использую RabbitMQ - возможно, что другие системы обмена сообщениями могут вести себя несколько иначе. Это работало в Celery 2.3.x и 2.4.x; Я не уверен, как далеко назад.

def get_celery_worker_status():
    ERROR_KEY = "ERROR"
    try:
        from celery.task.control import inspect
        insp = inspect()
        d = insp.stats()
        if not d:
            d = { ERROR_KEY: 'No running Celery workers were found.' }
    except IOError as e:
        from errno import errorcode
        msg = "Error connecting to the backend: " + str(e)
        if len(e.args) > 0 and errorcode.get(e.args[0]) == 'ECONNREFUSED':
            msg += ' Check that the RabbitMQ server is running.'
        d = { ERROR_KEY: msg }
    except ImportError as e:
        d = { ERROR_KEY: str(e)}
    return d

Ответ 2

Из документации сельдерея 4.2:

from your_celery_app import app


def get_celery_worker_status():
    i = app.control.inspect()
    stats = i.stats()
    registered_tasks = i.registered()
    active_tasks = i.active()
    scheduled_tasks = i.scheduled()
    result = {
        'stats': stats,
        'registered_tasks': registered_tasks,
        'active_tasks': active_tasks,
        'scheduled_tasks': scheduled_tasks
    }
    return result

конечно, вы можете/должны улучшить код с обработкой ошибок...

Ответ 3

Следующие работали для меня:

import socket
from kombu import Connection

celery_broker_url = "amqp://localhost"

try:
    conn = Connection(celery_broker_url)
    conn.ensure_connection(max_retries=3)
except socket.error:
    raise RuntimeError("Failed to connect to RabbitMQ instance at {}".format(celery_broker_url))

Ответ 4

Чтобы проверить то же самое с помощью командной строки в случае, когда сельдерей работает как демон,

  • Активируйте virtualenv и перейдите в каталог, где "приложение"
  • Теперь запустите: celery -A [app_name] status
  • Покажет, есть ли сельдерей или нет, плюс нет. узлов онлайн

Источник: http://michal.karzynski.pl/blog/2014/05/18/setting-up-an-asynchronous-task-queue-for-django-using-celery-redis/

Ответ 5

Один из способов проверить, отвечает ли какой-либо работник, - отправить широковещательную рассылку "ping" и вернуться с успешным результатом по первому ответу.

from .celery import app  # the celery 'app' created in your project

def is_celery_working():
    result = app.control.broadcast('ping', reply=True, limit=1)
    return bool(result)  # True if at least one result

Это передает "пинг" и будет ждать до одной секунды для ответов. Как только поступит первый ответ, он вернет результат. Если вы хотите, чтобы результат False быстрее, вы можете добавить аргумент timeout чтобы сократить время ожидания до отказа.