Celery and RabbitMQ

@celery_app.task(bind=True, default_retry_delay=10, max_retries=100, soft_time_limit=120, time_limit=240)
def search_and_run_extractor(self, resource_host, index, limit=-1, sid=None, counter=0):
    _scroll_timeout = '30m'
    _scroll_size = settings.EXTRACTOR_ELASTIC_SCROLL_SIZE
    _indexes = list(index)
    index = _indexes[-1]
    try:
        es = get_elastic(settings.ELASTIC_HOSTS)
        if sid is not None:
            # если sid указан, то делаем выборку по нему.
            data = es.scroll(scroll_id=sid, scroll=_scroll_timeout)
        else:
            # это первый запуск, нужно инициализировать скролл.
            if not es.indices.exists(index=index):
                # если индекса нет, то нам не с чем работать.
                raise StopScrolling(f"Index {index} not found!")
            # инициализируем скрол
            search_body = get_search_body(resource_host)
            data = es.search(index=index, scroll=_scroll_timeout, size=_scroll_size, body=search_body)
 
        # обновляем ID скрола
        sid = data['_scroll_id']
        hits_count = len(data['hits']['hits'])
        if hits_count > 0:
            counter += hits_count
            if 0 < limit <= counter:
                raise StopScrolling('Limit')
            # обрабатываем пачку данных
            process_hits(data['hits']['hits'])
            # кладем эту же задачу обратно в очередь, добавляя к параметрам sid, чтобы продолжить скроллинг после обработки запущенной пачки
            search_and_run_extractor.delay(resource_host, _indexes, limit=limit, sid=sid, counter=counter)
            return f'OK. {counter}'
        else:
            raise StopScrolling('End of objects!')
    except (SoftTimeLimitExceeded, elastic_exceptions.ElasticsearchException) as e:
        send_log(f"SFE {type(e).__name__}::{e.args}", task_id=resource_host, resource_host="extractor",
                 level=logging.ERROR, job_name="search_for_extract", index=index)
        search_and_run_extractor.delay(resource_host, _indexes, limit=limit, sid=sid, counter=counter)
        return 'Retry'
    except StopScrolling as e:
        send_log(f"SFE {type(e).__name__}::{e.args}", task_id=resource_host, resource_host="extractor",
                 level=logging.ERROR, job_name="search_for_extract", index=index)
        if len(_indexes) > 1:
            _indexes.pop()
            search_and_run_extractor.delay(resource_host, _indexes, limit=limit, sid=None, counter=0)
        return str(e.args)
    except Exception as e:
        send_log(f"SFE {type(e).__name__}::{e.args}", task_id=resource_host, resource_host="extractor",
                 level=logging.ERROR, job_name="search_for_extract", index=index)
        return 'Error'

CELERY_QUEUES = (
    ...
    Queue('extractors', Exchange('extractors'), routing_key='extractors', queue_arguments={'x-max-priority': 10}),
    Queue('scroll', Exchange('scroll'), routing_key='scroll', queue_arguments={'x-max-priority': 10}),
    ...
)
 
CELERY_DEFAULT_QUEUE = 'default'
CELERY_DEFAULT_EXCHANGE = 'default'
CELERY_DEFAULT_ROUTING_KEY = 'default'
 
CELERY_TASK_ROUTES = {
   ...
    'api.tasks.extraction.exctract': {'queue': 'extractors'},
    'api.tasks.extraction.search_and_run_extractor': {'queue': 'scroll'},
   ...
}