Celery, RabbitMQ удаляет работника из списка потребителей во время выполнения им задач

Я запустил свой celery worker, который использует RabbitMQ в качестве брокера, следующим образом:

celery -A my_app worker -l info -P gevent -c 100 --prefetch-multiplier=1 -Q my_app

Тогда у меня есть задача, которая выглядит примерно так:

@shared_task(queue='my_app', default_retry_delay=10, max_retries=1, time_limit=8 * 60)
def example_task():
    # getting queryset with some filtering
    my_models = MyModel.objects.filter(...)

    for my_model in my_models.iterator():
        my_model.execute_something()

Иногда эта задача может быть завершена менее чем за минуту, а иногда, во время высокой нагрузки, для ее завершения требуется более 5 минут.

Основная проблема заключается в том, что RabbitMQ постоянно удаляет моего работника из списка потребителей. Это выглядит действительно случайным. Из-за этого мне приходится перезапускать worker снова.

Workers также начинает выдавать такие ошибки:

SSLEOFError: EOF occurred in violation of protocol (_ssl.c:2396)

Иногда возникают следующие ошибки:

consumer: Cannot connect to amqps://my_app:**@example.com:5671/prod: SSLEOFError(8, 'EOF occurred in violation of protocol (_ssl.c:997)').

Couldn't ack 2057, reason:"RecoverableConnectionError(None, 'connection already closed', None, '')"

Я пытался добавить --without-heartbeat, но это ничего не дает.

Как решить эту проблему? Иногда выполнение моих задач занимает более 30 минут, и я не могу постоянно отслеживать, были ли рабочие выгнаны из rabbitmq.

Вернуться на верх