Java8并行流写WordCount，并不简单

/**
* 用于存储流的中间数据
*/
public class WordCounterTuple {
	private int counter; //用于记录当前词数量
	private boolean lastSpace;//标记上一个词是否是空格

	public WordCounterTuple( boolean lastSpace,int counter) {
		super();
		this.counter = counter;
		this.lastSpace = lastSpace;
	}
    // 省略getter，setter
}

import java.util.function.BiFunction;
	/**
	 * 针对每个char，进行处理
	 * 只有，当前char不是空格，上一个char是空格，计数才加1；
	 * 之后以当前字符，作为上一个字符，
	 * 注这里仅仅标记是否是空格
	 */
public class WordCountAccumulator 
         implements BiFunction<WordCounterTuple,Character,WordCounterTuple>{
		@Override
		public WordCounterTuple apply( 
                   WordCounterTuple lastWordCounterTuple,
                   Character currentChar) {
			if(Character.isWhitespace(currentChar)){
				return new WordCounterTuple(true,lastWordCounterTuple.getCounter());
			}else{
				return lastWordCounterTuple.isLastSpace()
						?new WordCounterTuple(false,lastWordCounterTuple.getCounter()+1)
						:new WordCounterTuple(false,lastWordCounterTuple.getCounter());
			}
		}
}

import java.util.function.BinaryOperator;
/**
*合并器，将多个流的的结果进行累加
*/
public class WordCountCombiner implements BinaryOperator<WordCounterTuple>{
	@Override
	public WordCounterTuple apply(WordCounterTuple t, WordCounterTuple u) {		
		return new WordCounterTuple(t.isLastSpace(),t.getCounter()+u.getCounter());
	}
}

/**
* 使用流的reduce方法执行
*/
public  static int countWords(Stream<Character> stream) {
		WordCounterTuple  wordCounter = stream.reduce(
				new WordCounterTuple( true,0),
				new WordCountAccumulator(),
				new WordCountCombiner());
			System.out.println(wordCounter.getCounter());
			return wordCounter.getCounter();
}

public class WordCount {
	//Taylor Swift《...Ready For It?》
	static final String SENTENCE_36 = "Knew he was a killer first time that I saw him "+//11
"　Wonder how many girls he had loved and left haunted "+//10
"　But if he's a ghost then I can be a phantom "+//11
"　Holdin' him for ransom ";//4

	public static void main(String[] args) {
//		//串行流
//		//将String，映射为Character流
		Stream<Character> charStream=IntStream.range(0,SENTENCE_36.length())
.mapToObj(SENTENCE_36::charAt);
		WordCount.countWords(charStream);
	}
}

/**
* 统计字符的并行方法。
*/
public  static int countWordsParallel(Stream<Character> stream) {
		WordCounterTuple  wordCounter = stream.parallel().reduce(
				new WordCounterTuple( true,0),
				new WordCountAccumulator(),
				new WordCountCombiner());
			System.out.println(wordCounter.getCounter());
			return wordCounter.getCounter();
}

import java.util.Spliterator;
import java.util.function.Consumer;
/**
 *	对字符进行分割的方式
 */
public class WorkCountSpliterator implements Spliterator<Character>{

	private String needSpliterator;
	private int currentCharAt;

	public WorkCountSpliterator(String needSpliterator) {
		super();
		this.needSpliterator = needSpliterator;
	}

	@Override
	public boolean tryAdvance(Consumer<? super Character> action) {
		action.accept(needSpliterator.charAt(currentCharAt++));
		return currentCharAt<needSpliterator.length();
	}

	@Override
	public Spliterator<Character> trySplit() {
		int remainSize=needSpliterator.length()-currentCharAt;
		if(remainSize<30){//剩余字符串小于30，进行串行处理，不再生产子拆分器
			return null;
		}
		for(int splitpos=currentCharAt+remainSize/2
				;splitpos<needSpliterator.length()
				;splitpos++){//采用折半搜索
			if(Character.isWhitespace(needSpliterator.charAt(splitpos))){//如果是空格进行拆分
				Spliterator<Character> subCountSpliterator= 
                 new WorkCountSpliterator(needSpliterator.substring(currentCharAt,splitpos)); 
				currentCharAt=splitpos;//向前推进缩小范围
				System.out.println("拆分了："+subCountSpliterator);
				return subCountSpliterator;
			}
		}
		return null;
	}

	@Override
	public String toString() {
		return "WorkCountSpliterator [needSpliterator="
              + needSpliterator + ", currentCharAt=" + currentCharAt + "]";
	}

	@Override
	public long estimateSize() {
		return needSpliterator.length()-currentCharAt;
	}

	@Override
	public int characteristics() {
		return ORDERED | SIZED | SUBSIZED | NONNULL | IMMUTABLE;
	}
}

public static void main(String[] args) {
		//并行流
		WorkCountSpliterator spliter=new WorkCountSpliterator(SENTENCE_36);
		Stream<Character> stream = StreamSupport.stream(spliter, true);
		WordCount.countWordsParallel(stream);
	}

public interface Spliterator<T> {

  //T 遍历的元素的类型

  boolean tryAdvance(Consumer<? super T> action);

   //类似Iterator ，如果还有元素返回true

  Spliterator<T> trySplit();//因为它可以把一些元素划出去分给第二个 Spliterator

   long estimateSize();//估计还剩下多少元素要遍历，约准确划分越均匀

   int characteristics();//对接口的特性描述，详见表格1

}