Узкое место в приложениях Stream вызывает потерянные сообщения

Question

• Сервер Spring Cloud Data Flow (SCDF) для Cloud Foundry v1.4.x
• Плитка службы RabbitMQ, предназначенная для передачи сообщений

Развернутый поток данных Spring CloudПоток имеет процессор, который может создавать исходящие сообщения быстрее, чем нисходящий процессор или приемник могут обрабатывать входящие сообщения.Это вызывает узкое место в транспорте RabbitMQ, что в конечном итоге приводит к потере сообщений.

В нашей частной облачной среде наш сервисный тайл Rabbit имеет настройки по умолчанию max-length=1000 и max-length-bytes=1000000.В настоящее время мы не можем изменить эти настройки для увеличения любого из этих мощностей.

Мы попытались установить значение prefetch для приложения-потребителя (я полагаю, что значение будет deployer.<appname>.rabbit.bindings.consumer.prefetch=10000), что, по-видимому, фактически увеличивает способность приложения-потребителя принимать больше сообщений за более короткий период времени., но это будет эффективно только в ограниченных сценариях.Если у нас есть чрезвычайно большой объем данных, проходящих через поток, мы все еще можем столкнуться с ограничением, когда сообщения отбрасываются.В приведенном выше примере мы, похоже, увеличиваем емкость потребляющего приложения с 1000 до 11000, установив предварительную выборку.

Мы также попытались использовать службу автоматического масштабирования, чтобы мы могли увеличить количество активных экземпляров потребляющего приложения, что также может явно увеличить его емкость.Это, однако, также похоже на решение проблемы с помощью пластыря, а не использования инфраструктуры и / или услуг, которые по своей природе способны гибко справляться с ожидаемыми объемными ожиданиями.Что если мы не знаем определенного времени дня, когда объемы будут значительно увеличиваться, и что, если объем увеличивается с такой скоростью, что пороговые значения ЦП в настройке автоматического масштабирования не могут достаточно быстро идти в ногу с активными экземплярами, чтобы избежатьпотерянные сообщения?

• мы не пытались настроить сервис RabbitMQ для гарантии доставки.Основываясь на документации, кажется, что легче сказать, когда сообщение не было доставлено, чем сделать доставку достоверной.мы не знаем, является ли это хорошим жизнеспособным вариантом, и ищем совета.
• мы не пробовали реализовывать какие-либо ограничения в самих наших потоковых приложениях.мы не знаем, является ли это хорошим жизнеспособным вариантом, и ищем совета.
• мы не пробовали привязывать приложения к DLQ или переставлять сообщения, которые не обрабатываются.мы не знаем, является ли это хорошим жизнеспособным вариантом, и ищем совета.
• мы не пытались привязать сервер SCDF к нашему собственному экземпляру службы Rabbit за пределами фрагментов службы Cloud Foundry.Теоретически это будет экземпляр RabbitMQ, который будет иметь больший контроль над глубиной очереди и ограничением размера байта, где мы можем установить их для более легкой обработки ожидаемых нагрузок.
• мы не пробовали альтернативный транспортный механизм, такой как Кафка.мы не знаем, является ли это хорошим жизнеспособным вариантом, и ищем совета.

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

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

Какой-нибудь совет от сообщества, или от Главных людей по этому поводу?

Answer 1

Channing

Спасибо за предоставление стольких деталей, вопросов и за ваш интерес как к Spring Cloud Stream, так и к SCDF, но я надеюсь, вы понимаете, что на самом деле это не вопрос для SO, так как в нем так многопеременные, на которые он не может иметь ответа и которые больше подходят для обсуждения какого-либо типа.Возможно, запрос функции в GitHub для любого из упомянутых проектов, и мы можем обсудить это там.В любом случае, я сделаю все возможное, чтобы это не осталось без ответа.

То, о чем вы спрашиваете, это обратное давление, и это действительно очень важный вопрос.Однако должно быть понимание того, что Spring Cloud Stream и впоследствии SCDF решили поддерживать несколько систем / протоколов обмена сообщениями (через связующие), чтобы соединять микро-сервисы вместе, а не создавать свои собственные.И не все системы / протоколы обмена сообщениями поддерживают обратное давление, и когда-то они предоставляют различные механизмы для его достижения, что затрудняет / делает невозможным предоставление какой-либо общей абстракции на уровне структуры.

Таким образом, это становится скорее дискуссией по архитектуре / дизайну, в которой я был бы более рад участвовать, но мне нужно больше контекста.Например, в контексте RabbitMQ, один из способов может быть для производителя опросить размер очереди (RabbitAdmin.queueProperties(queue)) и прекратить публикацию, если она превышает некоторый порог.Но, как я уже сказал, есть много других хитростей и способов сделать что-то, и нам определенно понадобится больше контекста.

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

Надеюсь, это поможет.,.