d'Ivan

как настроить консьюмер IBMMQ так, чтобы он забирал из очереди не все сообщения - без ограничений - которые ему хочет выдать сервер (менеджер), а как-то дозированно: по 1, 2, N и только после обработки уже вычитанных вычитывал следующие?

Зачем?
Используй СУБД, считывая записи с LIMIT, OFFSET

Для "общения" серверов между собой можно использовать WebSocket, gRPC, HTTP и много чего ещё. Если они в одной подсети, то это значительно упрощает их взаимодействие.
Сервер в роли диспетчера может отправлять задачи и контролировать их выполнение: отправить задачу с её идентификатором, а при получении ответа записать в СУБД, что задача Х выполнена. Там же и проверить, что если все задачи подсерверов выполнены, то перейти к следующему этапу.

У приложения потребителя очереди должен иметься менеджер соединения с RabbitMQ. Как только соединение было установлено, запускается поток с веб-сервером и health-check. У потребителя очереди - основой поток, у веб-сервера - дочерний. Каждый работает независимо друг от друга. Для менеджера соединения с RabbitMQ может использоваться функция обратного вызова в библиотеке.
Таким образом, при вызове /health (или /health/live) всегда возвращается статус 200.
При внезапном разрыве соединения с RabbitMQ менеджер соединения приложения штатным образом пытается восстановить соединение (скажем, до N раз каждую секунду) и если облом, то выходим из приложения с ненулевым статусом (допустим, 1). Если приложение падает, то вместе с ним и health-check, что вызовет триггер по перезапуску приложения у процесса, который ведет мониторинг приложения.
Мониторинг самого RabbitMQ обычно ведется отдельными от приложения средствами. RabbitMQ может запускаться как on premises, так и в SaaS (допустим, CloudAMQP).

1) Насколько правильно обрабатывать в одной очереди и загрузку файла и парсинг самого файла?
Если произойдёт какая-то ошибка при чтении файла

Пока я не вижу смысла в дополнительной очереди. При скачивании нужно определять код статуса и если возвращает 20x, то можно приступать к чтению. Если 40x, то это баг клиента. Если 50x или сбой связи, то можно сохранить в таблице с временем последней попытки, чтобы можно было выбирать когда нужно провести следующую попытку.
Если файл скачан без ошибок и он цел, то проблем с чтением файла быть не должно, при условии соблюдения формата файла. А если скачан частично при коде 20x, то повторная обработка ничего не даст.
Скачанный файл можно удалять, если импорт окончен. А при сбое можно заново не скачивать - экономия времени.
Смену статуса импорта сделать сразу по окончанию обработки.

2) Запись сообщений в лог происходит в отдельной очереди, а для смены статуса импорта желательно отправлять сообщения в отдельную очередь или можно прямо здесь выполнять код для смены статуса?

не понял почему нужна очередь для лога.

только изображения отправляются для обработки в одельной очереди. Я думаю, что это сильно напрягает саму очередь

сами изображения отправляются или что-то другое? Если первое, то это нелогично и это действительно нагружает очередь. Зачем все товары слать в очередь? Разве в очереди импорта нельзя считывать всё с файла и записывать сразу в СУБД? Это ведь никого задерживать не будет.

Получается, что все товары висят в одной очереди и ждут когда их проверят.

Нет, очередь нужна для обработки без ожидания. А для хранения используется СУБД с пометкой последней проверки и, возможно, временем, когда товар надо снова обновить. Какой-то периодический процесс может опрашивать таблицу: у кого протухли товары, отзовись!

Crash dump is being written to: /var/log/rabbitmq/erl_crash.dump..

Вот там ответ и кроется.

Вообще, AMQP - непростой протокол и в нам есть много взаимодействий по сети. У разных реализаций протокола могут быть разные особенности, даже в пределах того же языка программирования.
Если отправка по интернету, то вообще все может быть менее предсказуемо и соединения могут отваливаться спонтанно, что может приводить к "заморозке" или частым переподключениям.
Не предоставлено никакой информации об:
используемых версиях библиотек и RabbitMQ, рантайма, конфигураций сети, ... как и самого кода.

Уверен, что есть множество реализаций AMQP и для Python. У каждой из них должная быть своя документация.
При инициализации проекта можно создать соединение или даже лучше пул соединений, используя которое/ый в некотором издателе, отправлять в нужную очередь.

Если ответ клиенту на его запрос выдается не сразу, а с задержкой в связи с ограничениями из-за API третьей стороны, то логично ввести систему заявок. При запросе регистрируют заявку клиента, заносят данные в СУБД, кладут в очередь и сразу выдается ответ с номером заявки и ожидаемым временем готовности. Когда обработчик очереди берет задачу с учетом задержки 5 секунд, получает ответ от API и заносит ответ в готовой для последующего чтения форме в СУБД для соответствующего номера заявки. Независимо от обработчика очереди, клиенты обращаются за получением готовых результатов. И если таковые готовы (проверка по колонке в таблице), выдается ответ. Если еще не готов, выдается следующее ожидаемое окно времени. И т.д., до лимита проверок.

Если же перед ответом дожидаться пока пройдет 5 секунд, то такая система быстро захлебнется при наплыве запросов и будут исчерпаны ресурсы. Поэтому RPC не рассматриваю как подходящий вариант работы.

Написано же:

17:10:49.061 [error] ERROR: distribution port 25672 in use by another node: rabbit@DESKTOP-PS7PKS9
ERROR: distribution port 25672 in use by another node: rabbit@DESKTOP-PS7PKS9

По какой-то причине порт уже занят.

А в чём проблема отправлять из сервис_1 в брокер и идентификатор пользователя и идентификатор сервиса (экземпляра), к которому он подключен? Тогда каждый экземпляр сервис_2 будет знать кому предназначена команда. Не вижу смысла плодить очереди на каждого пользователя (и тем более заниматься их управлением) и даже для каждого экземпляра сервис_1.
Так что я за более простой вариант:

Или это все одна очередь и каждое сообщение должно быть вычитано каждым инстансом сервиса_1 а затем отфильтровано?

Можно сделать такой вариант запуска задач:
Планировщик задач запускает программу, проверяющую какие-то условия и когда получено решение о выполнении задачи, отправляет в очередь указание с необходимыми для выполнения аргументами и затем программа сразу завершается. При такой схеме не будет висеть процесс программы долгое время, а консюмеры будут обрабатывать задачи асинхронно и поочередно.
Но для такого малого количества задач затея сомнительна.

Если я отправлю в очередь 'send_push' сообщение '{post_id: 123}': , его получит консьюмер, выберет из БД 1тыс пользователей

А что мешает консьюмеру брать из БД по 1000 в цикле?

По окончанию регистрации необходимо отправить уведомление, поэтому логично сделать это через брокер сообщений (RabbitMQ и др.).
Служба Уведомления представляет из себя потребителя очереди и рассылает письма асинхронно относительно регистрации.
Сам транспорт менее важен чем то когда событие будет обработано.
Транспортом может быть и HTTP, и WebSocket.