Блог

У меня есть решение, которое состоит из 3 служб, есть 2 очереди для передачи сообщений из одной службы в другую, и оно использует библиотеку EasyNetQ C # для подключения к RabbitMQ.

Это схема архитектуры:

1. Получение данных из сети TCP (выполнение некоторой обработки) отправка в очередь A
2. Чтение из очереди A (выполнение некоторой обработки) отправка в очередь B
3. Чтение из очереди B, отправка обработанных данных в сторонний API

Сервисы 1 и 2 являются старыми сервисами, в то время как 3 был представлен недавно.

Дело в том, что 1 и 2 обрабатывают по одному журналу за раз, поэтому шаблон публикации / подписки имеет смысл, т.е. служба 1 получает одно сообщение и помещает это одно сообщение в очередь A для дальнейшей обработки, но Служба 3 должна отправлять как можно больше сообщений стороннему API, поскольку API может принимать до 1000 сообщений в одном HTTP-запросе POST.

Я думал о введении таймера, например, каждые X минут таймер запускает событие для чтения до 1000 сообщений из очереди и отправки их через HTTP в сторонний API, но, похоже, я не могу найти способ вручную прочитать N количество сообщений из очереди.

На данный момент я публикую сообщение из службы 2 в очередь B следующим образом:

_bus.Publish(myObject);

Затем в 3-м сервисе у меня есть подписка

_bus.SubscribeAsync<XQueueDTO>("x_queue_processor", ProcessMessageAsync);

но эта подписка всегда будет считывать по одному сообщению за раз из очереди.

Итак, каков был бы наилучший способ справиться с этим?

Вы можете изменить BasicQos настройки, чтобы позволить пользователю прочитать 1000 сообщений. Если это не решает вашу проблему, вы можете читать сообщения одно за другим, и когда их количество достигнет 1000, вы сможете запустить логику сервиса 3.

Правка # 1:

Ссылка:https://www.rabbitmq.com/tutorials/tutorial-two-dotnet.html