在Java中对单个流的元素执行多个不相关的操作

Question

如何对单个流的元素执行多个不相关的操作？

假设我有一个由文本组成的List<String>。列表中的每个字符串可能包含也可能不包含某个单词，它表示要执行的操作。比方说：

• 如果字符串包含'of'，则该字符串中的所有单词都必须计算在内
• 如果字符串包含 'for'，则必须返回第一次出现 'for' 之后的部分，从而生成包含所有子字符串的 List<String>

当然，我可以这样做：

List<String> strs = ...;

List<Integer> wordsInStr = strs.stream()
    .filter(t -> t.contains("of"))
    .map(t -> t.split(" ").length)
    .collect(Collectors.toList());

List<String> linePortionAfterFor = strs.stream()
    .filter(t -> t.contains("for"))
    .map(t -> t.substring(t.indexOf("for")))
    .collect(Collectors.toList());

但随后列表将被遍历两次，如果strs 包含大量元素，这可能会导致性能下降。

是否可以在不遍历列表两次的情况下以某种方式执行这两个操作？

Answer 1

如果您想要一个单程 Stream，那么您必须使用自定义 Collector（可以并行化）。

class Splitter {
  public List<String> words = new ArrayList<>();
  public List<Integer> counts = new ArrayList<>();

  public void accept(String s) {
    if(s.contains("of")) {
      counts.add(s.split(" ").length);
    } else if(s.contains("for")) {
      words.add(s.substring(s.indexOf("for")));
    }
  }

  public Splitter merge(Splitter other) {
    words.addAll(other.words);
    counts.addAll(other.counts);
    return this;
  }
}
Splitter collect = strs.stream().collect(
  Collector.of(Splitter::new, Splitter::accept, Splitter::merge)
);
System.out.println(collect.counts);
System.out.println(collect.words);

Answer 2

这是从不同方面解决 OP 的答案。首先，让我们看一下迭代列表/集合的快/慢。以下是我机器上的性能测试结果：

当：字符串列表长度 = 100，线程数 = 1，循环数 = 1000，单位 = 毫秒

---

OP：0.013

接受的答案：0.020

通过计数器函数：0.010

---

当：字符串列表的长度 = 1000_000，线程数 = 1，循环数 = 100，单位 = 毫秒

---

操作：99.387

接受的答案：89.848

通过计数器功能：59.183

---

结论：性能提升的百分比很小甚至更慢（如果字符串列表的长度很小）。通常，减少由更复杂的收集器加载到内存中的列表/集合的迭代是错误的。你不会得到太多的性能改进。如果存在性能问题，我们应该寻找其他地方。

这是我使用工具Profiler进行性能测试的代码：（我不打算在这里讨论如何进行性能测试。如果您对测试结果有疑问，可以使用任何您相信的工具再做一次）

@Test
public void test_46539786() {
    final int strsLength = 1000_000;
    final int threadNum = 1;
    final int loops = 100;
    final int rounds = 3;

    final List<String> strs = IntStream.range(0, strsLength).mapToObj(i -> i % 2 == 0 ? i + " of " + i : i + " for " + i).toList();

    Profiler.run(threadNum, loops, rounds, "OP", () -> {
        List<Integer> wordsInStr = strs.stream().filter(t -> t.contains("of")).map(t -> t.split(" ").length).collect(Collectors.toList());
        List<String> linePortionAfterFor = strs.stream().filter(t -> t.contains("for")).map(t -> t.substring(t.indexOf("for")))
                .collect(Collectors.toList());

        assertTrue(wordsInStr.size() == linePortionAfterFor.size());
    }).printResult();

    Profiler.run(threadNum, loops, rounds, "Accepted answer", () -> {
        Splitter collect = strs.stream().collect(Collector.of(Splitter::new, Splitter::accept, Splitter::merge));
        assertTrue(collect.counts.size() == collect.words.size());
    }).printResult();

    final Function<String, Integer> counter = s -> {
        int count = 0;
        for (int i = 0, len = s.length(); i < len; i++) {
            if (s.charAt(i) == ' ') {
                count++;
            }
        }
        return count;
    };

    Profiler.run(threadNum, loops, rounds, "By the counter function", () -> {
        List<Integer> wordsInStr = strs.stream().filter(t -> t.contains("of")).map(counter).collect(Collectors.toList());
        List<String> linePortionAfterFor = strs.stream().filter(t -> t.contains("for")).map(t -> t.substring(t.indexOf("for")))
                .collect(Collectors.toList());

        assertTrue(wordsInStr.size() == linePortionAfterFor.size());
    }).printResult();
}

Answer 3

您可以为此使用自定义收集器并且只迭代一次：

 private static <T, R> Collector<String, ?, Pair<List<String>, List<Long>>> multiple() {

    class Acc {

        List<String> strings = new ArrayList<>();

        List<Long> longs = new ArrayList<>();

        void add(String elem) {
            if (elem.contains("of")) {
                long howMany = Arrays.stream(elem.split(" ")).count();
                longs.add(howMany);
            }
            if (elem.contains("for")) {
                String result = elem.substring(elem.indexOf("for"));
                strings.add(result);
            }

        }

        Acc merge(Acc right) {
            longs.addAll(right.longs);
            strings.addAll(right.strings);
            return this;
        }

        public Pair<List<String>, List<Long>> finisher() {
            return Pair.of(strings, longs);
        }

    }
    return Collector.of(Acc::new, Acc::add, Acc::merge, Acc::finisher);
}

用法是：

Pair<List<String>, List<Long>> pair = Stream.of("t of r m", "t of r m", "nice for nice nice again")
            .collect(multiple());

Answer 4

如果你想通过一个列表有 1 个流，你需要一种方法来管理 2 个不同的状态，你可以通过将 Consumer 实现到新类来做到这一点。

    class WordsInStr implements Consumer<String> {

      ArrayList<Integer> list = new ArrayList<>();

      @Override
      public void accept(String s) {
        Stream.of(s).filter(t -> t.contains("of")) //probably would be faster without stream here
            .map(t -> t.split(" ").length)
            .forEach(list::add);
      }
    }

    class LinePortionAfterFor implements Consumer<String> {

      ArrayList<String> list = new ArrayList<>();

      @Override
      public void accept(String s) {
        Stream.of(s) //probably would be faster without stream here
            .filter(t -> t.contains("for"))
            .map(t -> t.substring(t.indexOf("for")))
            .forEach(list::add);
      }
    }

    WordsInStr w = new WordsInStr();
    LinePortionAfterFor l = new LinePortionAfterFor();

    strs.stream()//stream not needed here
        .forEach(w.andThen(l));
    System.out.println(w.list);
    System.out.println(l.list);