Как использовать потоки для обработки фрагментов в потоке записей JPA ⇐ JAVA

[Anonymous]

В Spring Boot 2.7 очень легко выполнить потоковую передачу результатов запроса к базе данных и их пакетную обработку с помощью JDBC:

Код: Выделить всё

   try (PreparedStatement ps = conn.prepareStatement(MY_SQL)) {
ps.setFetchSize(100);
ResultSet rs = ps.executeQuery();
do {
List chunk = new ArrayList();
while (rs.next() && (rowsReadInThisBatch < myBatchSize)) {
MyEntity entity = new MyEntity();
myEntity.setSomeCol(rs.getLong("some_col");
myEntity.setSomeCol2(rs.getLong("some_col2");
chunk.add(entity)
rowsReadInThisBatch++;
}
// lost more code.
process(chunk);
writeBackToDB(cunk);
// lots more code
} while (rowsReadInThisBatch == myBatchSize);

Таблица базы данных содержит > 1 миллиона строк, поэтому мы не можем просто считать их все в один объект из 1 миллиона строк, поскольку наши производственные серверы микросервисов имеют только 1 ГБ ОЗУ. В приведенном выше примере у нас есть только записи myBatchSize в оперативной памяти, и нам нужно только выполнить пакетную вставку total/myBatchSize в базу данных (а не одну вставку для каждой записи, что приведет к 1 миллиону обращений к базе данных).
Чтобы сделать это с потоками в JPA, не существует эквивалента rs.next(). Вместо этого вам придется использовать лямбда-выражения и потоки Java.
Есть ForEach:

Код: Выделить всё

    try(Stream myStream = postRepository.streamByCreatedOnSince(yesterday)) {
myStream.forEach(
....
);
}

Однако forEach чрезвычайно ограничен, поскольку вы не можете использовать нефинальные переменные для подсчета количества обработанных фрагментов и т. д.
Как мне преобразовать мои старые добрые циклы do и циклы while в потоки Java/лямбда-материалы, чтобы читать и обрабатывать фрагменты, а не обрабатывать каждый по отдельности без разделения на фрагменты? В идеале не использовать встроенный лямбда-код, который сложно тестировать отдельно (и трудно понять для экспертов, не являющихся экспертами по потокам Java)?
Суть в том, что потоки Java не имеют концепции счетчика или фрагментации.
В этой статье: https://www.baeldung.com/java-stream-batch-processing есть несколько очень эзотерических решений, требующих дополнительных библиотек, которых мы хотим избежать. Кроме того, неясно, считывают ли их решения весь поток в память, а затем разбивают его на куски или пакеты, чего мы хотим избежать из-за очень большого набора данных.

Подробнее здесь: https://stackoverflow.com/questions/796 ... of-records