Сергей

Нужно понять, что это за процесс, это может быть служба, которая перезапускается автоматически после завершения. На сколько я помню, у netstat есть флаг -b, который выводить имя процесса (только делает это на другой строке почему-то).

По номерам портов видно, что это другие подключения, рассматривали такой вариант, что RabbitMQ их всё-таки закрывает, но приложение успевает переподключиться?

Если очень надо, чтобы "таймауты" делал RabbitMQ, то в ограниченной форме на помощь может придти расширение DLX, но нужны ещё дополнительные очереди для отстойников. RabbitMQ это универсальная очередь, он не может решать микро-требования для каждого варианта использования.

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

Если клиент не закрывает соединение с конкретным инстансом (или если закрыл, то ответ не должен получить), то посмотрите в сторону RPC, вот с помощью чего оно работает, вот примеры.
Если результат требует много вычислительных ресурсов и повторять нежелательно, то либо создавать очереди под каждого клиента, либо где-то кэшировать результат (например, в СУБД).

Дополню по плагину, сам не использовал, но по описанию и коду можно сказать как оно работает.
(1) Когда в такой обменник публикуется сообщение:
- Сохраняется связка <routing_key>=<value> [code]
- Дальше маршрутизация происходит так же как и у Direct обменника [code]
(2) Когда к такому обменнику цепляется очередь:
- По ключу связки ищется значение из внутренней базы
- Сообщение с последним значение публикуется в очередь

То есть, если Вам нужно сохранять состояние для N объектов через такой механизм, то для получаения придётся каждый раз делать N биндингов очереди на обменник, затем N раз убирать этот биндинг. А так же не забыть повесить ограничение на длину очереди, так как никому реальные сообщение не требуются.

Как вариант можете форкнуть этот обменник и допилить его так, чтобы при публикации сохранялись только последние состояния (без дальнейшей публикации как Direct), а при подписке на обменник RabbitMQ засылал бы сразу все состояния :-)

Тут всё зависит от задачи и терминологии. Например, "пользователь" может быть как какой-то объект системы, который не имеет прямого коннекта к RabbitMQ и отделен слоем другого ПО, так и быть непосредственно подключённым к нему. В первом случае на уровне RabbitMQ нужно будет иметь возможность различать типы промежуточного ПО (например, отправка сообщений по почте, по телеге и т.п.), возможно их группы (например, шардинг по регионам), а конкретный ID пользователя RabbitMQ знать совсем тут не нужно и он может быть частью тела сообщения. Во втором случае на уровне RabbitMQ уже нужно будет различать конкретных пользователей и ID уже должен быть частью метаданных, видных RabbitMQ.
Дальше, если перейти от термина "пользователь" к термину "клиент" (например), то в первом случае количество клиентов будет равно количеству инстансов промежуточного ПО (например, 1000), а во втором случае - количеству конечных пользователей (например, 1 млн). И тут можно сказать, что для каждого клиента (или группы равнозначных воркеров) нужно создавать очередь. В первом случае их будет тысяча, а во втором - миллион.

И да, очереди у RabbitMQ не самые лёгкие, по моему опыту (на версии 3.6-3.7) 10 тыс. очередей занимали около 800 МБ памяти.

Exchange это объект сервера, клиент лишь должен знать его имя (строку) для отправки сообщения. Клиент может создавать новые эксченжи, для чего нужно предоставить больше информации RabbitMQ (например, тип), в мануалах это как раз и используют чтобы упроситить и показать всю конфигурацию явно. В реальном продакшене, имхо, лучше чтобы вся инфраструктура обменников и очередей была предподготовленна отдельными утилитами, но для небольших сервисов это большой оверхед и проще из кода всё создавать. Я бы рекомендовал посмотреть ещё в сторону такого решения.

Как упомянул Иван, Вам надо начать контролировать успешное получение сообщения RabbitMQ, для чего нужно включить специальный режим у канала отправки. Скорость отправки в таком режиме конечно будет намного ниже, чтобы это как-то улучшить (при условии что неважен порядок сообщений в очереди) можно:
1) Открыть несколько каналов и вести параллельную отправку по ним
2) Открыть канал без подтверждения отправки и с подтверждением, отправить в первый, к примеру, 999 сообщений, а затем во второй одно и ждать пока придёт подтверждение на последний (что-то может пойти не так, но выглядит рабочим).

Дополнительно можно попробовать Lazy режим у очереди.

Вот в документации это вполне себе описано.
Очередь может быть durable или transient. В первом случае брокер после перезапуска её восстановит, но это никак не связано с её содержимым. Сообщения восстановятся только если они были отправлены с флагом persistent. Метаданные объектов (очереди, обменники и т.п.) хранятся в Mnesia, тогда как сами сообщения отдельно и в другом формате.

1. exchange != queue, у Вас и то, и то, или же только очереди?
2. Чтобы что-то оценить, нужны сами параметры очередей и их значения, иначе только пальцем в небо.

Указанное решение в принципе рабочее (встречал такое в своей практике), но есть вероятность, что специализированный плагин будет работать эффективнее (но точно не скажу, просто предположение).
Плагин находится "под присмотром" официальных разработчиков, так что проблем с ним быть не должно. Инструкция по установке плагинов, собранный плагин есть, например, вот тут (в zip архиве файл .ez который понимает RabbitMQ).

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

В дополнение к предыдущему ответу можно ещё добавить вот такой механизм - Dead Letter Exchanges.