Сергей

Я не из мира PHP, но полагаю что можно попробовать basic_get вместо basic_consume, или попробовать опцию $nowait=true

По PHP возможно не подскажу, но всё же в таких ситуациях желательно публиковать и пример кода. По всей видимости, открывается канал (channel) на каждую публикацию сообщения, чего делать не стоит. Каналы нужны для параллельной работы через одно TCP подключение, так как AMQP протокол по своей натуре синхронный (следующую команду нельзя послать, если ожидается ответ от предыдущей). Если через один канал скорость публикации сообщений с подтверждением не устраивает, то можно:

• открыть несколько параллельных каналов,
  
• или использовать пакетную отправку, когда подтверждение приходит на каждые N запросов.
  

Была подобная проблема, сишный клиент в один поток не умел нормально в heartbeat. При отключении heartbeat'ов на уровне AMQP протокола были проблемы с "зависшими клиентами" на стороне RabbitMQ, когда приложение выходило непредвиденно. Настройка TCP keepalive помогла решить данную проблему.
Не знаю точно как работает современный php, но, если есть возможность держать отдельный поток для поддержания AMQP коннекта с включённым heartbeat'ом, то это будет лучше, чем через TCP keepalive. Отсылать ack обратно лучше всё же после завершения обработки задачи.

Если я понимаю верно, то при создании очереди (queue_declare) в качестве имени указывается пустая строка, на что RabbitMQ создаёт для неё случайное имя и добавляет флаг exclusive, вследствие чего очередь удаляется при закрытии подключения клиентом. Попробуйте дать собственное имя для очереди и посмотрите параметры вызова этого метода.

Если хочется реализовать всё через очередь, где каждый элемент это товар, то как вариант создать временную очередь, заполнить её новыми товарами, а потом сообщить дальше её имя. Если интересно уведомление о завершении импорта также получить через RMQ, можно создать обменник, а основной процесс создаёт себе служебную очередь и подписывается на этот обменник.
Но, имхо, если импорт происходит в одном процессе, а генерация кэша по одному элементу это очень дешёвая операция, то вероятно не очень эффективно размещать каждый товар отдельно, лучше складывать их пачками, чтобы также пачку и обработать.

На уровне данных. Когда поднимается ResplicaSet в MongoDB, она понимает, что данные нужно дублировать на несколько нод. Когда создаётся кластер RabbitMQ (по факту это кластер Erlang-машин), то по всем машинам дублируются лишь метаданные (информация об обменниках, очередях и т.п.), но сами данные остаются жить на одной ноде (но прозрачно передаются/забираются, если клиент подключён к другой ноде кластера). У них есть статья в доке по отказоустойчивости к потери данных.

На уровне коннектов. Драйвер MongoDB открывает подключения до каждой ноды ReplSet (может быть лишь до какого-то их количества, не уверен точно) и, в зависимости от операции и её параметров, использует то или иное подключение. То есть поддержка ReplSet заложена на уровне драйвера. RabbitMQ работает на протоколе AMQP (нативный), который работает только с одной нодой брокера, скорее всего есть куча библиотек, которые обёртывают в себе подключение к нескольким серверам RabbitMQ (либо один коннект и другой поднимается по отказу первого, либо сразу несколько коннектов, которые дают балансировку и малый лаг при отказе одной ноды). Но, если нет либы для работы с несколькими нодами, то сделать её, думаю, особых проблем не составит.