Какие знаешь нюансы при чтении 20 млн записей из базы данных и отправки в Message Broker?

Нюансы при чтении 20 млн записей и отправке в Message Broker

Обработка больших объемов данных из БД с передачей в Message Broker — сложная задача, требующая тщательного проектирования. Рассмотрим критические аспекты.

1. Управление памятью

Критическая проблема: загрузка всех 20 млн записей в память приведет к OutOfMemoryError.

// ПЛОХО: загрузка всего в памяти
List<Record> allRecords = getAllRecords(); // OutOfMemoryError!
for (Record record : allRecords) {
    messageBroker.send(record);
}

// ХОРОШО: постраничная загрузка (pagination)
int pageSize = 10000;
for (int offset = 0; offset < totalCount; offset += pageSize) {
    List<Record> page = getRecordsByPage(offset, pageSize);
    for (Record record : page) {
        messageBroker.send(record);
    }
}

Оптимальный размер страницы: 5000-50000 записей в зависимости от размера объекта и доступной памяти.

2. Использование курсоров (Cursor-based pagination)

Лучше чем offset: использование курсоров для больших datasets:

// ХОРОШО: cursor-based подход
String cursor = null;
while (true) {
    Page<Record> page = getRecordsByCursor(cursor, pageSize);
    if (page.isEmpty()) break;
    
    for (Record record : page) {
        messageBroker.send(record);
    }
    
    cursor = page.getNextCursor();
}

Преимущества:

• Избегаем скип большого количества строк (OFFSET N в SQL дорогой)
• Стабильна при вставке новых данных во время обработки
• O(1) вместо O(n)

3. Batch обработка в Message Broker

Отправка по одной — неэффективно. Используйте batch:

// ПЛОХО: отправка по одной
for (Record record : records) {
    messageBroker.send(record);
}

// ХОРОШО: batch отправка
List<Record> batch = new ArrayList<>();
for (Record record : records) {
    batch.add(record);
    if (batch.size() >= batchSize) {
        messageBroker.sendBatch(batch);
        batch.clear();
    }
}
if (!batch.isEmpty()) {
    messageBroker.sendBatch(batch);
}

Параметры batch: 100-1000 записей за один вызов.

4. Управление соединениями с БД

Connection pool критичен для долгой обработки:

// ПЛОХО: одно соединение (может timeout)
Connection conn = dataSource.getConnection();
for (int offset = 0; offset < total; offset += pageSize) {
    // длительная операция, соединение может выйти из строя
}

// ХОРОШО: переиспользование пула
try (Connection conn = dataSource.getConnection()) {
    for (int offset = 0; offset < total; offset += pageSize) {
        ResultSet rs = executeQuery(conn, offset, pageSize);
        // обработка
    }
}

Конфигурация:

• Timeout для соединения: 30-60 секунд
• Pool size: зависит от параллельности (8-20 соединений типично)
• Idle timeout: 10-15 минут

5. Обработка сетевых сбоев

Retry логика для Message Broker:

// ХОРОШО: экспоненциальная задержка
int maxRetries = 3;
int delayMs = 1000;

for (int retry = 0; retry < maxRetries; retry++) {
    try {
        messageBroker.send(batch);
        break;
    } catch (BrokerException e) {
        if (retry < maxRetries - 1) {
            Thread.sleep(delayMs);
            delayMs *= 2; // exponential backoff
        } else {
            logger.error("Failed to send batch after retries", e);
            throw e;
        }
    }
}

6. Дедупликация и идемпотентность

Problem: при сбое может быть дубликат отправки:

// ХОРОШО: использование idempotency key
public void processAndSend(Record record) {
    String idempotencyKey = UUID.randomUUID().toString();
    Message msg = new Message(
        record,
        idempotencyKey
    );
    messageBroker.send(msg);
    // Broker проверит: такой ключ уже обработан?
}

7. Мониторинг прогресса

Отслеживание обработки больших объемов:

// ХОРОШО: логирование метрик
long startTime = System.currentTimeMillis();
int processed = 0;
int failed = 0;

for (Record record : recordsIterator) {
    try {
        messageBroker.send(record);
        processed++;
    } catch (Exception e) {
        failed++;
        logger.error("Failed to process record: " + record.getId(), e);
    }
    
    if (processed % 100000 == 0) {
        long elapsed = System.currentTimeMillis() - startTime;
        double rate = (processed * 1000.0) / elapsed;
        logger.info(String.format(
            "Processed: %d, Failed: %d, Rate: %.0f rec/sec",
            processed, failed, rate
        ));
    }
}

8. Параллельная обработка

Multi-threaded подход для увеличения пропускной способности:

// ХОРОШО: асинхронная обработка
ExecutorService executor = Executors.newFixedThreadPool(4);
List<Future<?>> futures = new ArrayList<>();

for (Record record : records) {
    futures.add(executor.submit(() -> {
        try {
            messageBroker.send(record);
        } catch (Exception e) {
            logger.error("Error", e);
        }
    }));
}

for (Future<?> future : futures) {
    future.get(); // ждем завершения
}

Осторожно: необходимо управлять rate limiting и не перегружать Broker.

9. Обработка исключений и восстановление

Dead Letter Queue для неудачных сообщений:

// ХОРОШО: отправка в DLQ
try {
    messageBroker.send(record);
} catch (Exception e) {
    messageBroker.sendToDLQ(record, e);
    logger.warn("Message sent to DLQ", e);
}

10. Оптимизация SQL запросов

Используйте projection вместо загрузки всех полей:

// ПЛОХО: все поля
select * from records where ...

// ХОРОШО: только нужные поля
select id, name, status from records where ...

Чеклист решения

• Pagination: используй cursor-based, не offset
• Batch отправка: 100-1000 записей за раз
• Connection pool: правильная конфигурация
• Retry logic: экспоненциальная задержка
• Мониторинг: логируй метрики каждые 100k записей
• Асинхронность: параллельная обработка с контролем
• DLQ: обработка ошибок
• Идемпотентность: уникальные ключи
• Memory: стриминг, не загрузка всего
• Testing: тест на prod-подобном объеме данных

При таком подходе 20 млн записей будут обработаны эффективно за часы, не дни.