Давайте на мгновение согласимся с тем, что реализация RPC поверх очередей сообщений (например, RabbitMQ) не является ужасной идеей.

это совсем не страшно! это распространено и рекомендуется во многих ситуациях, а не только при интеграции с устаревшей версией.

... хорошо, теперь к вашему актуальному вопросу :)

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

Ваш запрос и ответ должны содержать две ключевые части информации:

• a correlation-id
• очередь reply-to

Эти биты информации позволят вам сопоставить исходный запрос и ответ.

Прежде чем отправить запрос

пусть ваш запрашивающий код создаст для себя эксклюзивную очередь. Эта очередь будет использоваться для получения ответов.

создайте новый идентификатор корреляции — обычно это GUID или UUID, чтобы гарантировать уникальность.

При отправке запроса

Прикрепите созданный вами идентификатор корреляции к свойствам сообщения. есть свойство correlationId, которое вы должны использовать для этого.

сохраните идентификатор корреляции со связанной функцией обратного вызова (обработчиком ответов) для запроса где-то внутри кода, выполняющего запрос. вам понадобится это, когда придет ответ.

также прикрепите имя созданной вами монопольной очереди к свойству replyTo сообщения.

со всем этим вы можете отправить сообщение через rabbitmq

при ответе

код ответа должен использовать поля correlationId и replyTo исходного сообщения. так что обязательно захватите их

ответ должен быть отправлен непосредственно в очередь replyTo. не используйте стандартную публикацию через биржу. вместо этого отправьте ответное сообщение непосредственно в очередь, используя функцию «отправить в очередь» любой библиотеки, которую вы используете, и отправьте ответ непосредственно в очередь replyTo.

не забудьте также включить correlationId в ответ. это важная часть, чтобы ответить на ваш вопрос

при обработке ответа

Код, выполнивший первоначальный запрос, получит сообщение из очереди replyTo. затем он вытащит correlationId из свойств сообщения.

используйте идентификатор корреляции, чтобы найти метод обратного вызова для запроса... код, который обрабатывает ответ. передайте сообщение этому методу обратного вызова, и все готово.

детали реализации

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

большинство хороших библиотек RabbitMQ для любого данного языка будут иметь встроенный запрос/ответ. Если у вас нет, вы можете поискать другую библиотеку. Если вы не пишете библиотеку на основе шаблонов поверх протокола AMQP, вам следует искать библиотеку, в которой реализованы общие шаблоны.

Если вам нужна дополнительная информация о шаблоне запроса/ответа, включая все детали, которые я предоставил здесь (и многое другое), ознакомьтесь со следующими ресурсами:

• Мой собственный электронный курс / электронная книга RabbitMQ Patterns
• Учебники по RabbitMQ
• Шаблоны корпоративной интеграции — обязательно купите книгу, чтобы получить полное описание/шаблон реализации. стоит иметь эту книгу

Если вы работаете в Node.js, я рекомендую использовать библиотеку wascally, которая включает в себя Request Функция /Reply, которая вам нужна. Для Ruby проверьте bunny. Для Java или .NET посмотрите на некоторые из многочисленных реализаций служебной шины. В .NET я рекомендую NServiceBus или MassTransit.

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

Как было предложено в комментариях, прямой ответ-на кажется подходящим способом. Я задокументировал здесь все варианты, которые я пробовал, прежде чем остановиться на этом.

Я написал пакет npm amq.rabbitmq.reply-to.js что:

• Использует прямой ответ — функция, которая позволяет клиентам RPC (запрос/ответ) с дизайном, аналогичным показанному в руководстве 6 (https://www.rabbitmq.com/direct-reply-to.html).), чтобы не объявлять очередь ответов для каждого запроса.

• Создает эмиттер событий, в котором ответы rpc будут публиковаться по коэффициенту корреляции, как предложено https://github.com/squaremo/amqp.node/issues/259#issuecomment-230165144

Применение:

const rabbitmqreplyto = require('amq.rabbitmq.reply-to.js');

const serverCallbackTimesTen = (message, rpcServer) => {
    const n = parseInt(message);
    return Promise.resolve(`${n * 10}`);
};

let rpcServer;
let rpcClient;
Promise.resolve().then(() => {
    const serverOptions = new rabbitmqreplyto.RpcServerOptions(
    /* url */ undefined, 
    /* serverId */ undefined, 
    /* callback */ serverCallbackTimesTen);

    return rabbitmqreplyto.RpcServer.Create(serverOptions);
}).then((rpcServerP) => {
    rpcServer = rpcServerP;
    return rabbitmqreplyto.RpcClient.Create();
}).then((rpcClientP) => {
    rpcClient = rpcClientP;
    const promises = [];
    for (let i = 1; i <= 20; i++) {
        promises.push(rpcClient.sendRPCMessage(`${i}`));
    }
    return Promise.all(promises);
}).then((replies) => {
    console.log(replies);
    return Promise.all([rpcServer.Close(), rpcClient.Close()]);
});

//['10',
//  '20',
//  '30',
//  '40',
//  '50',
//  '60',
//  '70',
//  '80',
//  '90',
//  '100',
//  '110',
//  '120',
//  '130',
//  '140',
//  '150',
//  '160',
//  '170',
//  '180',
//  '190',
//  '200']