自定义 Java 8 收集器

Question

我想了解如何实现自定义收集器。比如说，我需要做一些词汇分析，例如字母频率映射，以及将2个结果合并为一个结果的能力。

class CharHistogram implements Collector<String, Map<Character, Integer>, Map<Character, Integer>> {



    public static CharHistogram toCharHistogram(){
        return new CharHistogram();
    }

    @Override
    public Supplier<Map<Character, Integer>> supplier() {
        SysOut.print("supplier invoked");
        return HashMap::new;
    }

    @Override
    public BiConsumer<Map<Character, Integer>, String> accumulator() {
        SysOut.print("accumulator invoked");
        return (map, val) -> {
            SysOut.print(val +" processed");
            char[] characters = val.toCharArray();
            for (char character : characters) {
                int count = 1;
                if (map.containsKey(character)) {
                    count = map.get(character);
                    count++;
                }
                map.put(character, count);
            }
        };
    }

    @Override
    public BinaryOperator<Map<Character, Integer>> combiner() {
        SysOut.print("combiner invoked");
        return (map1, map2) -> {
            SysOut.print(map1+" merged to "+map2);
            map2.forEach((k, v) -> map1.merge(k, v, (v1, v2) -> v1 + v2));
            return map1;
        };
    }

    @Override
    public Function<Map<Character, Integer>, Map<Character, Integer>> finisher() {
        SysOut.print("finisher invoked");
        return Function.identity();
    }

    @Override
    public Set<java.util.stream.Collector.Characteristics> characteristics() {
        return Collections.unmodifiableSet(EnumSet.of(Characteristics.IDENTITY_FINISH, Characteristics.UNORDERED));
    }

}

客户端代码：

CharHistogram charStatsState = CharHistogram.toCharHistogram();

Map<Character, Integer> charCountMap = Arrays.asList("apple","orange","orange").stream().collect(charStatsState);
SysOut.print(charCountMap);
charCountMap = Arrays.asList("pears","pears","orange").stream().collect(charStatsState);
SysOut.print(charCountMap);

输出：

[main]: supplier invoked
[main]: accumulator invoked
[main]: combiner invoked
[main]: apple processed
[main]: orange processed
[main]: orange processed
[main]: {p=2, a=3, r=2, e=3, g=2, l=1, n=2, o=2}
[main]: supplier invoked
[main]: accumulator invoked
[main]: combiner invoked
[main]: pears processed
[main]: pears processed
[main]: orange processed
[main]: {p=2, a=3, r=3, s=2, e=3, g=1, n=1, o=1}

我没有看到组合器或完成器被调用，我认为这些需要正确设计才能实现我想要的结果。

我错过了什么吗？

编辑：

支持流和组合器的可能方法。下面的代码不起作用。

class CharStreamHistogram implements Function<String, Map<Character, Integer>>{

    private int totalCharactersRead;
    private Map<Character, Integer> histogram;

    public int getTotalCharactersRead() {
        return totalCharactersRead;
    }
    public Map<Character, Integer> getHistogram() {
        return histogram;
    }
    public void setHistogram(Map<Character, Integer> histogram) {
        this.histogram = histogram;
    }
    public void setTotalCharactersRead(int totalCharactersRead) {
        this.totalCharactersRead = totalCharactersRead;
    }

    public Map<Character, Integer> combine(Map<Character, Integer>  map2) {
        Map<Character, Integer> map1 = this.histogram;
        map2.forEach((k, v) -> map1.merge(k, v, (v1, v2) -> v1 + v2));
        return map2;
    }


    @Override
    public Map<Character, Integer> apply(String val) {
        char[] characters = val.toCharArray();
        totalCharactersRead += characters.length;
        for (char character : characters) {
            int count = 1;
            if (histogram.containsKey(character)) {
                count = histogram.get(character);
                count++;
            }
            histogram.put(character, count);
        }
        return histogram;
    }

} 

public static <T> Collector<T, ?, CharStreamHistogram> summarizeCharStream(
             CharStreamHistogram histogram) { //TODO: is this correct?
        Collector charStatsState = new Collector<String, CharStreamHistogram, CharStreamHistogram>() {

            @Override
            public Supplier<CharStreamHistogram> supplier() {
                return CharStreamHistogram::new;
            }

            @Override
            public BiConsumer<CharStreamHistogram, String> accumulator() {
                //TODO: What to do here?
                return null;
            }

            @Override
            public BinaryOperator<CharStreamHistogram> combiner() {
                BinaryOperator binaryOperator = (l, r) -> {
                    l.combine(r); //TODO: Something like this?
                };
                return binaryOperator;
            }

            @Override
            public Function<CharStreamHistogram, CharStreamHistogram> finisher() {
                //TODO: What to do here?
                return null;
            }

            @Override
            public Set<java.util.stream.Collector.Characteristics> characteristics() {
                return Collections.unmodifiableSet(EnumSet.of(Characteristics.UNORDERED));
            }
        };
        return charStatsState;
    }

- John Eipe

---

你的 Collector 不应该有任何状态。你的累加器对象应该包含 totalCharactersRead 和 map。 - Louis Wasserman

@LouisWasserman 谢谢。所以现在我会把它移除。但是我之后会再处理这个问题。 - John Eipe

Answer 1

你已经声明了 Characteristics.IDENTITY_FINISH - 这明确表示 finisher 不会被调用，而且只有在并行流的情况下才会调用 combiner。

- Eugene

---

我并没有真正理解 finisher() 的作用。上面的代码对于这个使用情况来说是正确的吗？ - John Eipe

是的，在收集器中除了状态之外，既不需要合并器也不需要完成器来满足这种用例。 - Louis Wasserman

但我需要将两个结果合并的能力。我该怎么做？ - John Eipe

基本上这就是我正在尝试的：（1）一种收集统计数据的方法，（2）一种将新统计数据添加到现有统计数据中的方法。 - John Eipe

@John 当你想学习如何编写自定义收集器时，仍应遵循最佳实践来学习。首先，没有理由自己实现“Collector”。您可以简单地调用“Collector.of（...）”，传递三个或四个函数和可选特性。当省略finisher时，“IDENTITY_FINISH”将自动添加。此外，您在组合器函数中展示了对“Map.merge”的了解，那么为什么不在累加器中也使用它呢？而且，不要为了“for”循环而将整个字符串复制到“char []”数组中。使用“string.chars().forEach（...）”即可。 - Holger