java - 我可以通过作为流项目字段的 int 值的总数来限制 Java Stream 长度吗？

Question

我知道 limit 方法可以将 Java Stream length 限制为指定值。

但是有没有一种方法可以通过作为流项目字段的int 值的总数来限制Stream 的长度？

流由 List 组成。

要求是:

• 该方式需要能够并行执行进程
• 该方式需要创建Stream对象。
• 如果可能，我更喜欢使用 Java8。

现在我通过下面的方式来实现。

1. 使用 for 语句并计算总数。
2. 然后当总数达到限制值时中断for语句。

但这种方式不能满足要求。

如果给出limitByTotalIntOf方法，我想要的理想方式是:

class Item { int value; }
List<Item> listOfItems = createListOfItems();

int limitValue = 10000;
listOfItems.parallelStream()
    .map(item -> /* set value */)
    .limitByTotalIntOf(limitValue, item -> item.value)
    .map(item -> /* do something */);

目前的方式是:

int total = 0;
for(Item item: listOfItems) {
    int v = /* set value */;
    total += v;
    if(total < limitValue) break;
    /* do something */
}

有什么办法可以满足要求吗？

Answer 1

假设您的 /* do something */ 是一项真正受益于并行处理的昂贵操作，最好的解决方案是将受影响项目的选择与其实际处理分开。

最好按顺序进行选择，类似于您已有的循环:

List<Item> selectedItems = listOfItems;
int total = 0;
for(int ix = 0, num = listOfItems.size(); ix < num; ix++) {
    total += listOfItems.get(ix).getValue();
    if(total > limitValue) {
        selectedItems = listOfItems.subList(0, ix);
        break;
    }
}
selectedItems.parallelStream()
// your actual heavy operation

请注意，鉴于当前的实现，子列表上的并行流比在整个列表上的并行流上使用 limit(…) 更有效。

---

并行流处理的工作原理是拆分整个元素序列，在最好的情况下就在中间，然后拆分剩余的 block ，直到有足够的 block 让每个 CPU 核心保持忙碌。自然地，这种策略不太适合从序列开头到元素的选择，条件是包含所有受影响的元素。

但为了完整起见，可以并行执行此操作，尽管不太可能从中受益。您需要非常多的元素，并且必须选择很大一部分元素。

int[] values = selectedItems.parallelStream().mapToInt(Item::getValue).toArray();
Arrays.parallelPrefix(values, Integer::sum);
int index = Arrays.binarySearch(values, limitValue);
index = index < 0? -index-1: index+1;
listOfItems.subList(0, index).parallelStream()
// your actual heavy operation