1. Введение: Проблемы архитектуры повторной передачи больших объемов данных (Backfill)
В современных распределенных системах архитектура надежность потоков данных напрямую связана с жизнеспособностью сервиса. Особенно временные ряды (TimeSeries) данных, возникающие в Интернете вещей (IoT), финансовом секторе или системах мониторинга в реальном времени часто требуют более десятка тысяч операций в секунду с высокой пропускной способностью (Throughput). В таких условиях процесс повторной передачи (Backfill) неудачных данных без пропусков из-за сбоев системы, разрывов сети или временной недоступности downstream базы данных является весьма сложной задачей для бэкенд-инженеров.
Одной из распространенных ошибок, совершаемых многими разработчиками, является простое повторное отправление неудачных данных в цикле или их замещение в общую публичную очередь сообщений. Однако в среде крупномасштабной эксплуатации такой подход может вызвать цепную неисправность (Cascading Failure), которая приведет к остановке реального сервиса.
В этом техническом документе описывается опыт повышения надежности системы, стабильно и эффективно повторяющей миллионы накопленных временных рядов, сочетая высокоскоростную инфраструктуру обмена сообщениями NATS, высокопроизводительную неблокирующую структуру данных Java ConcurrentLinkedQueue и кастомный асинхронный пул потоков Spring (@Async), при этом надежно защищая реальный трафик и описывая стратегии оптимизации.
2. Ограничения существующей архитектуры и технические болевые точки (Pain Point)
Начальная система до повышения на ранних этапах имела простую структуру, в которой одновременно обрабатывались реалтайм трафик и трафик повторной передачи в одной очереди сообщений. Эта структура не проявила дефектов в маломасштабной тестовой среде, но в операционной среде момент срабатывания крупномасштабной работы по backfill вызвал три критических узких места.
2.1. Отключение реального сервиса из-за исчерпания ресурсов (Resource Exhaustion)
Запросы реальных пользователей и запросы на backfill делили ресурсы общего HTTP пула потоков и единой очереди сообщений. Вследствие этого, когда в систему поступало миллионы запросов для переноса данных, рабочие потоки повторной передачи захватывали все ресурсы CPU и памяти, а также доступные потоки. В результате ключевые запросы, которые должны были обрабатываться в реальном времени, не могли получить выделенные потоки, ожидая бесконечно, что приводило к катастрофе с истечением времени соединения (Connection Timeout).
Рисунок 1. Проблема исчерпания ресурсов, возникающая при совместном использовании одинаковых ресурсов для реальных запросов и запросов на повторную передачу.
2.2. Всплеск нагрузки в результате временной (Time-based) периодичности
При загрузке данных backfill в downstream базу данных (PostgreSQL) из очереди в начале использовалась логика, основанная на времени, например, "каждые 10 секунд запускать планировщик для очистки очереди". Это увеличивало непредсказуемость, когда количество поступающих данных колебалось. Если за определенные 10 секунд данные накапливались экспоненциально, то во время следующего планирования десятки тысяч запросов одновременно сбрасывались в базу данных, что приводило к перегрузке пула соединений и 100% загрузке процессора базы данных - «всплеск трафика (Spike)» происходил часто.
2.3. Расточительство ресурсов из-за блокировки потоков (Thread Blocking)
Во время проверки момента завершения повторной передачи данных backfill и активации финального процесса агрегации (Aggregation) использовался синхронный блокирующий механизм, такой как Thread.sleep() или CountDownLatch, под предлогом «ожидание завершения обработки всех данных». Это создавало ситуации, когда ожидающие потоки продолжали удерживать память и часто приводили к переключению контекста, что значительным образом снижало общую эффективность использования ресурсов системы.
3. Четыре стратегии повышения архитектуры для изоляции и оптимизации
Для преодоления трех вышеупомянутых ограничений была внедрена архитектура Bulkhead (габаритная изоляция), которая полностью отделяет общую структуру от уровня инфраструктуры, структуры данных приложения и модели потоков.
3.1. Изоляция слоя инфраструктуры и структур данных: NATS и ConcurrentLinkedQueue
Первым делом было осуществлено тщательное разделение маршрутизации трафика. Темы реального времени и темы, предназначенные только для обратной выгрузки, были разделены на уровне инфраструктуры (NATS), а внутри Java-приложения был зарегистрирован независимый буферный объект памяти @Bean для безопасного хранения только обратных данных.
@Configuration
public class QueueConfig {
@Bean
public Queue<TimeSeriesVo> backfillTimeSeriesVoQueue() {
// 다중 스레드 환경에서 Lock-Free 고성능을 보장하는 논블로킹 큐 선택
return new ConcurrentLinkedQueue<>();
}
}Ключевым моментом является выбор ConcurrentLinkedQueue. Обычные LinkedBlockingQueue или ArrayBlockingQueue требуют от потоков производителя (Producer) и потребителя (Consumer) получения блокировки (Lock) при доступе к очереди, что может приводить к узким местам в ожидании, когда наблюдается высокая конкуренция за ресурсы. В отличие от этого, ConcurrentLinkedQueue работает на основе алгоритма CAS (Compare-And-Swap), который является атомарной операцией на уровне CPU и не использует блокировки. Даже если потребитель NATS с невероятно быстрой скоростью обрабатывает десятки тысяч данных в секунду, он может безопасно и без блокировки получать данные в память, исполняя роль буферной зоны в взаимодействии с рабочими потоками.
Рисунок 2. Множество потоков потребителей NATS загружает данные в очередь без блокировки с использованием алгоритма CAS
3.2. Изоляция слоя потоков: @Async("backfillTimeSeriesPool")
Даже если данные были успешно получены через NATS и очередь памяти, если рабочие процессы, выполняющие фактическую бизнес-логику и загружающие данные в БД, используют общий пул потоков веб-сервера, проблема исчерпания ресурсов, упомянутая ранее, не будет решена.
Поэтому, воспользовавшись механизмом @Async в Spring, мы создали независимый пользовательский пул потоков, отвечающий только за задачи обратной выгрузки.
@Configuration
@EnableAsync
public class AsyncConfig {
@Bean(name = "backfillTimeSeriesPool")
public Executor backfillTimeSeriesPool() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5); // 기본 유지할 스레드 개수
executor.setMaxPoolSize(10); // 트래픽 폭발 시 최대 확장 스레드 개수
executor.setQueueCapacity(500); // Task 대기 큐 용량
executor.setThreadNamePrefix("backfill-task-");
executor.setRejectedExecutionHandler(
new ThreadPoolExecutor.CallerRunsPolicy());
executor.initialize();
return executor;
}
}Мы объявили такой изолированный пул потоков и внедрили его в асинхронные компоненты рабочей нагрузки, обрабатывающие обратные выгрузки.
@Component
public class BackfillWorker {
private final Queue<TimeSeriesVo> backfillTimeSeriesVoQueue;
private final TimeSeriesRepository timeSeriesRepository;
public BackfillWorker(
Queue<TimeSeriesVo> backfillTimeSeriesVoQueue,
TimeSeriesRepository timeSeriesRepository) {
this.backfillTimeSeriesVoQueue = backfillTimeSeriesVoQueue;
this.timeSeriesRepository = timeSeriesRepository;
}
@Async("backfillTimeSeriesPool")
public void processBackfill() {
// 공용 HTTP 스레드를 전혀 오염시키지 않고, 백필 전용 스레드(backfill-task-) 내에서만 안전하게 루프 구동
List<TimeSeriesVo> chunkList = new ArrayList<>();
while (!backfillTimeSeriesVoQueue.isEmpty()) {
TimeSeriesVo vo = backfillTimeSeriesVoQueue.poll();
if (vo != null) {
chunkList.add(vo);
// 3.3 섹션에서 서술할 크기 기반 청크 처리 로직이 여기에 위치함
if (chunkList.size() >= 1000) {
timeSeriesRepository.saveAll(chunkList);
chunkList.clear();
}
}
}
if (!chunkList.isEmpty()) {
timeSeriesRepository.saveAll(chunkList);
}
}
}Эта архитектура позволяет задачам обратной выгрузки, работающим в фоновом режиме в течение нескольких часов, полностью изолироваться, так что http-nio-потоки, обрабатывающие реальные запросы пользователей, могут поддерживать производительность на уровне меньше 0,1 секунды в спокойном режиме.
3.3. Управление потоком на стороне сервера (Flow Control): Извлечение на основе размера (Size-based)
Чтобы решить проблему бесконтрольной выгрузки базы данных в заданиях на основе времени, было изменено направление на метод Size-based Extraction (извлечение на основе размера), чтобы поддерживать оптимальную скорость в рамках предельной мощности системы.
Когда поток извлекает данные из очереди, он основывается исключительно на количестве данных (Chunk Size), а не на времени. Мы оцениваем оптимальный размер пакета, который база данных (PostgreSQL) может воспринять наиболее стабильно с точки зрения обновления индексов и нагрузки соединения, и непрерывно отправляем данные в БД частями указанного размера, как только очередь наполняется (poll()).
С помощью этого управления потоком мы обеспечили структурную стабильность, чтобы избежать худшего сценария, когда база данных может быть перегружена, даже если объем обратных данных резко увеличивается до сотен тысяч, и контролировать верхний предел нагрузки в пределах критической точки системы, что позволяет постепенно завершать массовую вставку (Bulk Insert).
Рисунок 3. Планирование на основе времени вызывает всплески нагрузки, но извлечение на основе размера поддерживает постоянную пропускную способность
3.4. Оптимизация асинхронных ресурсов: контрольный флаг и проверка завершения isEmpty() без блокировки
После того как данные обратной подгрузки будут полностью retransmitted, система должна выполнить агрегирующую операцию для обеспечения окончательной целостности данных. Мы отказались от старого метода блокировки потоков и объединили контрольный флаг и проверку состояния очереди для завершения полноценного асинхронного потока.
Когда сигнал завершения обратной подгрузки (End-of-Stream Flag), который сообщает о завершении передачи данных, достигает потребителя через протокол NATS, система переключает статусный флаг на COMPLETING. После этого рабочие потоки не засыпают и не ждут, а проверяют состояние главной очереди без блокировок.
public void checkAndTriggerAggregation() {
// 스레드를 재우는 오버헤드 없이, 락 프리로 큐의 공백 상태를 체크
if (isEndOfStreamSignalReceived && backfillTimeSeriesVoQueue.isEmpty()) {
// 대기 시간 0초 만에 즉시 최종 집계 프로세스 비동기 트리거
aggregationService.triggerFinalAggregation();
}
}Как только backfillTimeSeriesVoQueue.isEmpty() вернёт true, финальная логика агрегирования запустится без каких-либо задержек ЦП или бездействующих потоков.
Кроме того, из-за особенностей распределенной асинхронной среды, чтобы предотвратить некоторые состояния гонки (Race Condition), когда порядок 'запроса данных обратной подгрузки' и 'приема успешного ответа' меняется из-за сетевых задержек, мы добавили логику, которая сначала регистрирует конкретные нормальные события ответа в белый список (Whitelist) в памяти и выполняет их перекрестную проверку, чтобы уменьшить вероятность потери данных.
4. Заключение: урок, извлечённый из гармонии инфраструктуры и программного обеспечения
Самый ценный урок, полученный в рамках этого проекта по усовершенствованию, заключается в том, что "даже если мы внедрим высокопроизводительную распределенную инфраструктуру (NATS), если структуры данных и модели потоков внутри бэкенд-приложения будут недостаточными, узкое место изменит своё место и в конечном итоге вновь станет проблемой".
Разделение тем на уровне инфраструктуры, изоляция пользовательских потоков @Async через настройки Spring, буферизация без блокировок (Lock-Free) с помощью ConcurrentLinkedQueue и управление по размеру с использованием чанк-контроля образовали цепочку, которая в конечном итоге завершила настоящий большой поток данных.
Опыт улучшения архитектуры, который обеспечил идеальную целостность данных в условиях массовой повторной передачи, снижая при этом ресурсоемкость системы, стал основой для согласования компромиссов (Trade-off) в условиях ограниченных вычислительных ресурсов и обеспечения надежной стабильности системы, которая не колебалась в самых экстремальных условиях трафика.
Pancake Maker