好的,感谢所有的贡献。我将所说的内容结合起来,并实现了我需要的功能。也许实现过程会澄清我最初想要的。
我创建了两个类,RepositryCollector 和 RepositorySpliterator。
public class RepositoryCollector<T> implements Collector<T, Tuple2<Integer,List<T>>, Integer>{
private JpaRepository<T, ?> repository;
private int threshold;
public BinaryOperator<Tuple2<Integer, List<T>>> combiner() {
return (listTuple, itemsTuple) -> {
List<T> list = listTuple._2;
List<T> items = itemsTuple._2;
list.addAll(items);
int sum = listTuple._1 + itemsTuple._1;
if(list.size() >= this.threshold){
this.repository.save(list);
this.repository.flush();
list = new LinkedList<>();
}
return new Tuple2<>(sum, list);
};
}
}
我省略了收集器所需的其他函数,因为组合器中已经包含了所有相关信息。对于Spliterator也是如此。
public class RepositorySpliterator<T> implements Spliterator<T> {
private Slice<T> slice;
private Function<Pageable, Slice<T>> getSlice;
private Iterator<T> sliceIterator;
public RepositorySpliterator(Pageable pageable, Function<Pageable, Slice<T>> getSlice) {
this.getSlice = getSlice;
this.slice = this.getSlice.apply(pageable);
this.sliceIterator = slice.iterator();
}
@Override
public boolean tryAdvance(Consumer<? super T> action) {
if(sliceIterator.hasNext()) {
action.accept(sliceIterator.next());
return true;
} else if (slice.hasNext()) {
this.slice = getSlice.apply(slice.nextPageable());
this.sliceIterator = this.slice.iterator();
if(sliceIterator.hasNext()){
action.accept(sliceIterator.next());
return true;
}
}
return false;
}
public Stream<T> getStream(boolean parallel){
return StreamSupport.stream(this, parallel);
}
}
正如您所见,我添加了一个帮助函数来生成所需的流。也许这有点粗糙,但是...算了。
因此,现在我只需要在我的映射类中加入几行代码就可以开始了。
public void start(Timestamp startTimestamp, Timestamp endTimestamp) {
new RepositorySpliterator<>(
new PageRequest(0, 10000), pageable -> sourceRepository.findAllBetween(startTimestamp, endTimestamp, pageable))
.getStream(true)
.map(entity -> mapToTarget(endTimestamp, entity))
.collect(new RepositoryCollector<>(targetRepository, 1000));
}
映射器将从源中获取10000个实体,将它们倒入流池中,以便进行映射和存储。每当一个流用完新的实体时,就会获取一批新的实体,并将其馈送到同一个流池中。
如果我的实现中存在明显的错误,请随时评论并改进!