Mieliśmy podobny problem do rozwiązania. Chcieliśmy wziąć strumień, który był większy niż pamięć systemowa (iterując po wszystkich obiektach w bazie danych) i jak najlepiej losowo uporządkować kolejność - pomyśleliśmy, że byłoby dobrze zbuforować 10000 elementów i je losować.
Celem była funkcja, która przyjmowała strumień.
Spośród proponowanych tutaj rozwiązań wydaje się, że istnieje szereg opcji:
- Użyj różnych dodatkowych bibliotek innych niż java 8
- Zacznij od czegoś, co nie jest strumieniem - np. Listą dostępu swobodnego
- Miej strumień, który można łatwo podzielić w rozdzielaczu
Początkowo naszym instynktem było użycie niestandardowego kolektora, ale oznaczało to rezygnację z przesyłania strumieniowego. Powyższe niestandardowe rozwiązanie kolektora jest bardzo dobre i prawie go użyliśmy.
Oto rozwiązanie, które oszukuje, wykorzystując fakt, że Streams może dać ci, Iteratorktórego możesz użyć jako włazu ewakuacyjnego, abyś mógł zrobić coś więcej, czego strumienie nie obsługują. IteratorJest przekształcany z powrotem do strumienia za pomocą innej, Java 8 StreamSupportczary.
public class BatchingIterator<T> implements Iterator<List<T>> {
public static <T> Stream<List<T>> batchedStreamOf(Stream<T> originalStream, int batchSize) {
return asStream(new BatchingIterator<>(originalStream.iterator(), batchSize));
}
private static <T> Stream<T> asStream(Iterator<T> iterator) {
return StreamSupport.stream(
Spliterators.spliteratorUnknownSize(iterator,ORDERED),
false);
}
private int batchSize;
private List<T> currentBatch;
private Iterator<T> sourceIterator;
public BatchingIterator(Iterator<T> sourceIterator, int batchSize) {
this.batchSize = batchSize;
this.sourceIterator = sourceIterator;
}
@Override
public boolean hasNext() {
prepareNextBatch();
return currentBatch!=null && !currentBatch.isEmpty();
}
@Override
public List<T> next() {
return currentBatch;
}
private void prepareNextBatch() {
currentBatch = new ArrayList<>(batchSize);
while (sourceIterator.hasNext() && currentBatch.size() < batchSize) {
currentBatch.add(sourceIterator.next());
}
}
}
Prosty przykład użycia tego wyglądałby następująco:
@Test
public void getsBatches() {
BatchingIterator.batchedStreamOf(Stream.of("A","B","C","D","E","F"), 3)
.forEach(System.out::println);
}
Powyższe wydruki
[A, B, C]
[D, E, F]
W naszym przypadku chcieliśmy przetasować partie, a następnie zachować je jako strumień - wyglądało to tak:
@Test
public void howScramblingCouldBeDone() {
BatchingIterator.batchedStreamOf(Stream.of("A","B","C","D","E","F"), 3)
.map(list -> {
Collections.shuffle(list); return list; })
.flatMap(List::stream)
.forEach(System.out::println);
}
Wyprowadza coś takiego (jest losowy, więc za każdym razem inny)
A
C
B
E
D
F
Sekretem jest to, że zawsze istnieje strumień, więc możesz albo działać na strumieniu partii, albo zrobić coś z każdą partią, a następnie z flatMappowrotem do strumienia. Jeszcze lepiej, wszystkie powyższe tylko działa jako ostateczne forEachlub collectczy inne wyrazy kończące PULL dane przez strumień.
Okazuje się, że iteratorjest to szczególny rodzaj operacji kończącej na strumieniu i nie powoduje on uruchomienia całego strumienia i zapamiętania go! Podziękowania dla facetów z Java 8 za genialny projekt!