模拟hadoop的计算单词个数统计的helloword程序，计算给定语句的单词出现次数。进一步优化：按照次数进行排序后输出。
使用了parallel并行计算，三个参数的reduce进行缩减计算。三个参数时是最难以理解的。先来看其定义：
<U> U reduce(U identity,
BiFunction<U, ? super T, U> accumulator,
BinaryOperator<U> combiner)
分析下它的三个参数：
- identity: 一个初始化的值；这个初始化的值其类型是泛型U，与Reduce方法返回的类型一致；注意此时Stream中元素的类型是T，与U可以不一样也可以一样，这样的话操作空间就大了；不管Stream中存储的元素是什么类型，U都可以是任何类型，如U可以是一些基本数据类型的包装类型Integer、Long等；或者是String，又或者是一些集合类型ArrayList等；
- accumulator: 其类型是BiFunction，输入是U与T两个类型的数据，而返回的是U类型；也就是说返回的类型与输入的第一个参数类型是一样的，而输入的第二个参数类型与Stream中元素类型是一样的。
- combiner: 其类型是BinaryOperator，支持的是对U类型的对象进行操作；
代码如下：
//1. 给定一段英文句子，单词用空格间隔
String words = " guo xiu zhi guo map reduce guo hadoop java stream stream parallel guo guo";
//2.1 由于是多线程要使用 ConcurrentHashMap，HashMap不支持并发操作
//2.2 给定语句通过空格分割成数组，转换成普通Stream，再转换并行流
//2.3 并行流进行三参数的reduce计算。第一个参数是返回值，第二个参数是累加器，第三个参数是各个线程返回值的合并操作。
ConcurrentHashMap<String, Integer> reduceMap = Arrays.stream(words.split(" ")).parallel().reduce(new ConcurrentHashMap<String, Integer>(), (map, w) -> {
Integer num = map.get(w);
if (num == null) {//这个单词如果不存在于map中
map.put(w, 1);
} else {//这个单词如果存在于map中，数量+1，新值放入map
map.put(w, num + 1);
}
return map;
}, (m1, m2) -> {
m1.putAll(m2);//把并行计算的各个结果归并整合到m1中
return m1;
});
//根据单词的出现次数倒序排列后输出到List对象
List<Entry<String, Integer>> collect = reduceMap.entrySet().stream().sorted((o1, o2) -> o2.getValue() - o1.getValue()).collect(Collectors.toList());
//输出排序后的list数据
System.out.println("reduceMap = " + collect);
输出内容：
reduceMap = [guo=4, stream=2, reduce=1, =1, xiu=1, java=1, parallel=1, zhi=1, hadoop=1, map=1]
统计单词数量结果
但是我们发现数据偶尔会有丢失。对于parrallel之后元素数量不固定的原因，就是多线程有可能同时读取到相同的下标n然后同时赋值，这样就会出现元素缺失的问题。无非就是多个线程赋值可能同时操作同一个地址，后赋值的把先赋值的给覆盖掉了，才会出现这种问题。所以在使用paralleStream时不使用foreach、map操作，完成collect后的新Stream再使用。
什么方法可以防止并行流出现线程不安全操作？
-
【推荐方法】Java8 Stream的collect方法，就是收集Stream里的元素，返回List，Set或Map等，并且它是线程安全的。那就是最后调用collect(Collectors.tolist())，这种收集起来所有元素到新集合是线程安全的。在采用并行流收集元素到集合中时，由于我们使用了并发流要使用
Collectors.toConcurrentMap收集到ConcurrentMap，而不是使用普通的toMap()，toList()方法进行收集。
ConcurrentMap<String, Integer> toConcurrentMap = Arrays.stream(words.split(" ")).parallel().collect(Collectors.toConcurrentMap(s -> s, s -> 1,
(integer, integer2) -> integer + integer2));
List<Entry<String, Integer>> orderedCollection = toConcurrentMap.entrySet().stream().sorted((o1, o2) -> o2.getValue() - o1.getValue()).collect(Collectors.toList());
System.out.println("reduceMap = " + orderedCollection);
没有使用外部定义的集合，并且使用到了parallelStream并行处理的优势。输出的数据不再丢失
reduceMap = [stream=3, guo=2, zhi=2, reduce=1, java=1, xiu=1, parallel=1, hadoop=1, map=1]
- 给map操作加锁
Lock lock = new ReentrantLock();
ConcurrentHashMap<String, Integer> reduceMap = Arrays.stream(words.split(" ")).parallel().reduce(new ConcurrentHashMap<String, Integer>(), (map, w) -> {
lock.lock();
Integer num = map.get(w);
if (num == null) {//这个单词如果不存在于map中
map.put(w, 1);
} else {//这个单词如果存在于map中，数量+1，新值放入map
map.put(w, num + 1);
}
lock.unlock();
return map;
}, (m1, m2) -> {
//把并行计算的各个结果归并整合到m1中，m1和m2是一个对象，所以直接返回m1
System.out.println(Thread.currentThread().getName() + "," + (m1 == m2));
return m1;
});
- 抽取map操作片段为单独方法
getStringIntegerConcurrentHashMap，加上synchronized
ConcurrentHashMap<String, Integer> reduceMap = Arrays.stream(words.split(" ")).parallel().reduce(new ConcurrentHashMap<String, Integer>(), (map, w) -> {
return getStringIntegerConcurrentHashMap(map, w);
}, (m1, m2) -> {
//把并行计算的各个结果归并整合到m1中，m1和m2是一个对象，所以直接返回m1
System.out.println(Thread.currentThread().getName() + "," + (m1 == m2));
return m1;
});
public synchronized static ConcurrentHashMap<String, Integer> getStringIntegerConcurrentHashMap(ConcurrentHashMap<String, Integer> map, String w) {
Integer num = map.get(w);
if (num == null) {//这个单词如果不存在于map中
map.put(w, 1);
} else {//这个单词如果存在于map中，数量+1，新值放入map
map.put(w, num + 1);
}
return map;
}
