如何在Java 8 Stream上實現“分區”操作？通過分區，我的意思是將一個流分成給定大小的子流。它在某種程度上與Guava Iterators.partition（）方法相同，只是希望分區是延遲評估的Streams而不是List的。
3 回答
開滿天機
TA貢獻1786條經驗 獲得超13個贊
將任意源流劃分為固定大小的批次是不可能的，因為這會加重并行處理。并行處理時，您可能不知道拆分后的第一個子任務中有多少個元素，因此您無法為下一個子任務創建分區，直到完全處理第一個子任務。
但是，可以從隨機訪問中創建分區流List。此類功能在例如我的StreamEx圖書館中可用：
List<Type> input = Arrays.asList(...);
Stream<List<Type>> stream = StreamEx.ofSubLists(input, partitionSize);
或者，如果您真的想要流：
Stream<Stream<Type>> stream = StreamEx.ofSubLists(input, partitionSize).map(List::stream);
如果您不想依賴第三方庫，則可以ofSubLists手動實現這種方法：
public static <T> Stream<List<T>> ofSubLists(List<T> source, int length) {
if (length <= 0)
throw new IllegalArgumentException("length = " + length);
int size = source.size();
if (size <= 0)
return Stream.empty();
int fullChunks = (size - 1) / length;
return IntStream.range(0, fullChunks + 1).mapToObj(
n -> source.subList(n * length, n == fullChunks ? size : (n + 1) * length));
}
該實現看起來有些長，但是考慮了一些極端情況，例如接近MAX_VALUE列表大小。
如果您想為無序流提供并行友好的解決方案（因此您不必擔心將哪些流元素合并為單個批處理），則可以使用這樣的收集器（感謝@sibnick的啟發）：
public static <T, A, R> Collector<T, ?, R> unorderedBatches(int batchSize,
Collector<List<T>, A, R> downstream) {
class Acc {
List<T> cur = new ArrayList<>();
A acc = downstream.supplier().get();
}
BiConsumer<Acc, T> accumulator = (acc, t) -> {
acc.cur.add(t);
if(acc.cur.size() == batchSize) {
downstream.accumulator().accept(acc.acc, acc.cur);
acc.cur = new ArrayList<>();
}
};
return Collector.of(Acc::new, accumulator,
(acc1, acc2) -> {
acc1.acc = downstream.combiner().apply(acc1.acc, acc2.acc);
for(T t : acc2.cur) accumulator.accept(acc1, t);
return acc1;
}, acc -> {
if(!acc.cur.isEmpty())
downstream.accumulator().accept(acc.acc, acc.cur);
return downstream.finisher().apply(acc.acc);
}, Collector.Characteristics.UNORDERED);
}
用法示例：
List<List<Integer>> list = IntStream.range(0,20)
.boxed().parallel()
.collect(unorderedBatches(3, Collectors.toList()));
結果：
[[2, 3, 4], [7, 8, 9], [0, 1, 5], [12, 13, 14], [17, 18, 19], [10, 11, 15], [6, 16]]
這樣的收集器是完全線程安全的，并為順序流生成有序批。
如果要對每個批次應用中間轉換，則可以使用以下版本：
public static <T, AA, A, B, R> Collector<T, ?, R> unorderedBatches(int batchSize,
Collector<T, AA, B> batchCollector,
Collector<B, A, R> downstream) {
return unorderedBatches(batchSize,
Collectors.mapping(list -> list.stream().collect(batchCollector), downstream));
}
例如，通過這種方式，您可以即時對每個批次中的數字求和：
List<Integer> list = IntStream.range(0,20)
.boxed().parallel()
.collect(unorderedBatches(3, Collectors.summingInt(Integer::intValue),
Collectors.toList()));
千萬里不及你
TA貢獻1784條經驗 獲得超9個贊
正如Jon Skeet在其評論中所示，似乎不可能使分區變得懶惰。對于非延遲分區，我已經有了以下代碼：
public static <T> Stream<Stream<T>> partition(Stream<T> source, int size) {
final Iterator<T> it = source.iterator();
final Iterator<Stream<T>> partIt = Iterators.transform(Iterators.partition(it, size), List::stream);
final Iterable<Stream<T>> iterable = () -> partIt;
return StreamSupport.stream(iterable.spliterator(), false);
}
添加回答
舉報
0/150
提交
取消