So verwenden Sie Streams, um Blöcke im JPA-Datenstrom zu verarbeiten ⇐ Java

[Anonymous]

In Spring Boot 2.7 ist das Zurückstreamen von Ergebnissen aus einer Datenbankabfrage und deren stapelweise Verarbeitung mithilfe von JDBC sehr einfach:

Code: Select all

   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);

Die Datenbanktabelle hat > ​​1 Million Zeilen, daher können wir sie nicht einfach alle in ein einziges 1-Million-Zeilen-Objekt einlesen, da unsere Produktions-Microservice-Server nur über 1 GB RAM verfügen. Im obigen Beispiel haben wir immer nur myBatchSize-Datensätze im RAM und müssen nur total/myBatchSize-Batch-Einfügungen in die Datenbank durchführen (nicht eine Einfügung für jeden Datensatz, was zu 1 Million Roundtrips zur DB führen würde).
Um dies mit Streams in JPA zu tun, gibt es kein Äquivalent zu rs.next(). Stattdessen müssen Sie Lambdas und Java-Streams verwenden.
Es gibt ForEach:

Code: Select all

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

forEach ist jedoch äußerst eingeschränkt, da Sie keine nicht endgültigen Variablen verwenden können, um zu zählen, wie viele verarbeitet wurden usw.
Wie konvertiere ich meine einfachen alten do-Schleifen und while-Schleifen in Javas Streams/Lambda-Sachen, um Chunks einzulesen und Chunks zu verarbeiten und nicht jeden einzeln ohne Chunking zu verarbeiten? Idealerweise keinen Inline-Lambda-Code verwenden, der schwer separat zu testen ist (und für Nicht-Java-Streams-Experten schwer zu verstehen ist)?
Der Knackpunkt ist, dass Java-Streams nicht über das Konzept eines Zählers oder Chunking verfügen.
Dieser Artikel: https://www.baeldung.com/java-stream-batch-processing enthält einige sehr esoterische Lösungen, die zusätzliche Bibliotheken erfordern, die wir vermeiden möchten. Außerdem ist nicht klar, ob ihre Lösungen den gesamten Stream in den Speicher einlesen und ihn dann in Blöcke oder Stapel aufteilen, was wir aufgrund des sehr großen Datensatzes vermeiden möchten.