Когда издатель ожидает ответа на сообщение, как гарантировать, что он получит только релевантные ответы (на свои собственные сообщения) при его масштабировании?
У нас есть клиентский процесс, который публикует сообщение дляпроцесс сервера, чтобы ответить.Кроме того, у нас есть процесс «слушателя», который просто должен потреблять как вопросы, так и ответы, ничего не публикуя.Кроме того, процесс сервера может быть разбит на несколько в будущем, создавая каскад сообщений.Мы не можем использовать запрос / ответ, так как нам нужен слушатель, а затем снова, когда у нас будет каскад ... Кроме того, у нас будет несколько категорий вопросов / ответов, а запрос / ответ в EasyNetQ не поддерживает темы.
Наше решение с EasyNetQ было простой публикацией / подпиской на основе тем: клиент публикует в теме " question ", подписывается на " answer ", сервер подписывается на " question"и публикует на" answer", а слушатель просто подписывается на оба.
Проблема в том, что вы масштабируете клиента.Два его экземпляра теперь публикуют вопросы, но, поскольку они оба подписаны на одну тему " answer ", можно получить ответ на вопрос, опубликованный другим экземпляром, а не получить свой собственный.
Решение, которое мы нашли, заключается в том, чтобы клиент использовал очередь с уникальным именем при подписке на « answer » - таким образом, каждый клиент получит все ответы, и ему просто нужно игнорировать те, которыене его.Однако это решение имеет некоторые недостатки в производительности, а также приводит к тому, что очереди с уникальными именами накапливаются в RabbitMQ каждый раз, когда клиент падает (или просто перезапускается во время разработки и т. Д.).
Клиент, отправляющий объект msg:
string corrId = Guid.NewGuid().ToString();
// Register the corrId in a dictionary
//...
var myMessage = new MyMessage {correlationId =corrId, realMessage = msg};
easyNetQBus.Subscribe<MyMessage>("mqClient"+uniqueSuffix, HandleMsg, x => x.WithTopic("answer"));
easyNetQBus.Publish(myMessage, "question");
// In HandleMsg, we see if we have issued questions with the correlation id that came with the answer (lookup in the dictionary) and if not, ignore it
Сервер:
easyNetQBus.Subscribe<MyMessage>("mqServer", HandleMsg, x => x.WithTopic("question"));
// In HandleMsg, we publish the answer back to "answer" with the correlation id from the question
Есть ли другой шаблон, который мы должны использовать?Мы могли бы поместить в каждое сообщение уникальную тему / очередь для отправки ответа, но это усложняет жизнь слушателя и гибкость будущих участников каскада, о котором я упоминал ...