java - 与collect(supplier,accumulator,combiner)并行使用java流没有给出预期的结果

Question

我正在尝试查找给定字符串中的单词数。下面是它的顺序算法，效果很好。

public int getWordcount() {

        boolean lastSpace = true;
        int result = 0;

        for(char c : str.toCharArray()){
            if(Character.isWhitespace(c)){
                lastSpace = true;
            }else{
                if(lastSpace){
                    lastSpace = false;
                    ++result;
                }
            }
        }

        return result;

    }

但是，当我尝试使用 Stream.collect(supplier,accumulator,combiner) 方法“并行化”此方法时，我得到 wordCount = 0。我使用不可变类 (WordCountState) 只是为了维护字数统计的状态.

代码:

public class WordCounter {
    private final String str = "Java8 parallelism  helps    if you know how to use it properly.";

public int getWordCountInParallel() {
        Stream<Character> charStream = IntStream.range(0, str.length())
                                                .mapToObj(i -> str.charAt(i));

        WordCountState finalState = charStream.parallel()                                             
                                              .collect(WordCountState::new,
                                                        WordCountState::accumulate,
                                                        WordCountState::combine);

        return finalState.getCounter();
    }
}

public class WordCountState {
    private final boolean lastSpace;
    private final int counter;
    private static int numberOfInstances = 0;

public WordCountState(){
        this.lastSpace = true;
        this.counter = 0;
        //numberOfInstances++;
    }

    public WordCountState(boolean lastSpace, int counter){
        this.lastSpace = lastSpace;
        this.counter = counter;
        //numberOfInstances++;
    }

//accumulator
    public WordCountState accumulate(Character c) {


        if(Character.isWhitespace(c)){
            return lastSpace ? this : new WordCountState(true, counter);
        }else{
            return lastSpace ? new WordCountState(false, counter + 1) : this;
        }   
    }

    //combiner
    public WordCountState combine(WordCountState wordCountState) {  
        //System.out.println("Returning new obj with count : " + (counter + wordCountState.getCounter()));
        return new WordCountState(this.isLastSpace(), 
                                    (counter + wordCountState.getCounter()));
    }

我发现上述代码有两个问题:1. 创建的对象(WordCountState)数量大于字符串中的字符数量。2.结果始终为0。3. 根据累加器/消费者文档，累加器不应该返回 void 吗？即使我的累加器方法返回一个对象，编译器也不会提示。

有什么线索我可能偏离了轨道吗？

更新:使用的解决方案如下 -

public int getWordCountInParallel() {
        Stream<Character> charStream = IntStream.range(0, str.length())
                                                .mapToObj(i -> str.charAt(i));


        WordCountState finalState = charStream.parallel()
                                              .reduce(new WordCountState(),
                                                        WordCountState::accumulate,
                                                        WordCountState::combine);

        return finalState.getCounter();
    }

Answer 1

您始终可以调用方法并忽略其返回值，因此在使用方法引用时允许相同的操作是合乎逻辑的。因此，当需要消费者时，只要参数匹配，创建对非 void 方法的方法引用是没有问题的。

您使用不可变的 WordCountState 类创建的内容是一个归约操作，即它将支持这样的用例

Stream<Character> charStream = IntStream.range(0, str.length())
                                        .mapToObj(i -> str.charAt(i));

WordCountState finalState = charStream.parallel()
        .map(ch -> new WordCountState().accumulate(ch))
        .reduce(new WordCountState(), WordCountState::combine);

而 collect 方法支持可变缩减，其中容器实例(可能与结果相同)会被修改。

您的解决方案中仍然存在逻辑错误，因为每个 WordCountState 实例都以假设前面有空格字符开始，而不知道实际情况，也没有尝试在组合器中修复此问题。

仍然使用归约来解决和简化这个问题的方法是:

public int getWordCountInParallel() {
    return str.codePoints().parallel()
        .mapToObj(WordCountState::new)
        .reduce(WordCountState::new)
        .map(WordCountState::getResult).orElse(0);
}


public class WordCountState {
    private final boolean firstSpace, lastSpace;
    private final int counter;

    public WordCountState(int character){
        firstSpace = lastSpace = Character.isWhitespace(character);
        this.counter = 0;
    }

    public WordCountState(WordCountState a, WordCountState b) {
        this.firstSpace = a.firstSpace;
        this.lastSpace = b.lastSpace;
        this.counter = a.counter + b.counter + (a.lastSpace && !b.firstSpace? 1: 0);
    }
    public int getResult() {
        return counter+(firstSpace? 0: 1);
    }
}

如果您担心 WordCountState 实例的数量，请注意与您的初始方法相比，此解决方案未创建多少 Character 实例。

但是，如果您将 WordCountState 重写为可变结果容器，则此任务确实适合可变缩减:

public int getWordCountInParallel() {
    return str.codePoints().parallel()
        .collect(WordCountState::new, WordCountState::accumulate, WordCountState::combine)
        .getResult();
}


public class WordCountState {
    private boolean firstSpace, lastSpace=true, initial=true;
    private int counter;

    public void accumulate(int character) {
        boolean white=Character.isWhitespace(character);
        if(lastSpace && !white) counter++;
        lastSpace=white;
        if(initial) {
            firstSpace=white;
            initial=false;
        }
    }
    public void combine(WordCountState b) {
        if(initial) {
            this.initial=b.initial;
            this.counter=b.counter;
            this.firstSpace=b.firstSpace;
            this.lastSpace=b.lastSpace;
        }
        else if(!b.initial) {
            this.counter += b.counter;
            if(!lastSpace && !b.firstSpace) counter--;
            this.lastSpace = b.lastSpace;
        }
    }
    public int getResult() {
        return counter;
    }
}

请注意如何使用 int 一致地表示 unicode 字符，允许使用 CharSequence 的 codePoint() 流，这不仅是更简单，但也可以处理基本多语言平面之外的字符，并且可能更高效，因为它不需要装箱到 Character 实例。