• Stream API就是流操作api，Java中 流 都是指数据，尤其是大量数据，所以Stream API是为了方便处理大量数据的，也就是给大数据运算做基础的。

  • 注意区别于io流，io的流操作的数据是字节码或字符，主要是处理输入输出。

  • Stream api的流操作的数据是数组或集合，主要是数据的计算，把一个集合转变为另一个集合的过程中，进行一系列流水线式的操作，主要是数据的查找、过滤和映射等操作。

  • 集合是数据，流（Stream）是指计算 数据集合或数据数组的计算，称为Stream，也就是我们常说的流式计算

  • 相通于Hadoop框架的流式计算，map-reduce思想是一致的

image.png
注意：

  • stream api是基于lambda表达式的，也就是基于函数式编程的
  • stream不会自己存储数据，只是处理数据
  • stream不会改吧源数据对象，处理结束后会返回一个新生成的stream
  • stream操作是有延时的，意味着它们会等到·需要结果的时候才执行

    一、创建Stream

  1. //1. 通过Collection系列集合提供的 stream()或ParallelStream() 获取及河流
  2. List<String> list = new ArrayList<>();
  3. Stream<String> stream = list.stream();//非并行的流
  4. Stream<String> parallelStream = list.parallelStream();//并行的流
  6. //2. 通过Arrays的静态方法 stream()获取数组流
  7. Stream<Integer> stream1 = Arrays.stream(new Integer[10]);
  9. //3. 通过Stream类的静态方法of()
  10. Stream<Serializable> stream2 = Stream.of(1, 2, "initit.com");
  12. //4. 创建无限流
  13. //4.1 迭代方式
  14. Stream<Integer> stream3 = Stream.iterate(1001, x -> x + 2);
  15. /*
  16.         说明：
  17.          方法：iterate(final T seed, final UnaryOperator<T> f)
  18.          其中 UnaryOperator 接口如下：
  19.             @FunctionalInterface
  20.             public interface UnaryOperator<T> extends Function<T, T> {
  21.                 static <T> UnaryOperator<T> identity() {
  22.                     return t -> t;
  23.                 }
  24.             }
  25.          */
  27. //终止 测试
  28. stream3.limit(5).forEach(System.out::println);
  30. // 4.2 生成方式
  31. Stream<Double> stream4 = Stream.generate(Math::random);
  32. stream4.limit(100).forEach(System.out::println);
  33. //说明： generate(Supplier<T> s)
  34. //@FunctionalInterface
  35. //public interface Supplier<T> {
  36. //    T get();
  37. //}

二、中间操作

  • filter 过滤数据
  • limit 截断数据，不输出后边的数据
  • skip 跳过数据，不输出前边的数据
  • distinct 去重
  • map 映射
  • flatMap 平铺的映射，map和flatMap的区别等同于add和addAll的区别,flatMap是把多个流连成一个流，addAll是把多个集合连成一个集合
  • sorted 排序，默认排序
  • sorted(comparator)重写comparator方法，定制排序
  1. //模拟一个User对象
  2. class User {
  3.     private Integer id; //主键
  4.     private String name; //姓名
  5.     private Integer age;//年龄
  6.     private Double salary;//薪水
  8.     public User(String name, Integer age, Double salary) {
  9.         this.name = name;
  10.         this.age = age;
  11.         this.salary = salary;
  12.     }
  14.     //构造函数 方便测试
  15.     public User(String name, int age, double salary) {
  16.         this.name = name;
  17.         this.age = age;
  18.         this.salary = salary;
  19.     }
  21.     //后边要比对对象是否相同 需要重写equals和hasCode方法
  22.     @Override
  23.     public boolean equals(Object o) {
  24.         if (this == o) return true;
  25.         if (!(o instanceof User)) return false;
  26.         User user = (User) o;
  27.         return Objects.equals(getId(), user.getId()) &&
  28.             Objects.equals(getName(), user.getName()) &&
  29.             Objects.equals(getAge(), user.getAge()) &&
  30.             Objects.equals(getSalary(), user.getSalary());
  31.     }
  33.     @Override
  34.     public int hashCode() {
  35.         return Objects.hash(getId(), getName(), getAge(), getSalary());
  36.     }
  38.     //toString 方便控制台输出查看
  39.     @Override
  40.     public String toString() {
  41.         return "User{" +
  42.             "id=" + id +
  43.             ", name='" + name + '\'' +
  44.             ", age=" + age +
  45.             ", salary=" + salary +
  46.             '}';
  47.     }
  49.    //省略get set方法
  50. }
  52. //准备数据
  53. List<User> list = new ArrayList<User>() {
  54.     {
  55.         add(new User("张三", 18, 6000));
  56.         add(new User("张三", 18, 6000));
  57.         add(new User("张三", 18, 6000));
  58.         add(new User("李四", 20, 6000));
  59.         add(new User("张三", 50, 8000));
  60.         add(new User("编程指南", 30, 120000));
  61.         //为测试速度 添加一千万数据
  62.         //            for (int i = 0; i < 1000 * 10000; i++) {
  63.         //                add(new User(UUID.randomUUID() + "", 30, 120000));
  64.         //            }
  65.     }
  66. };
  69. Stream<User> stream = list.stream();
  71. @Test
  72. public void filter() {
  73.     //过滤 filter
  74.     stream.filter(user -> user.getAge() > 18).forEach(System.out::println);
  75.     // Stream<T> filter(Predicate<? super T> predicate);
  76.     //Predicate的方法 boolean test(T t);
  78.     //输出如下：
  79.     //User{id=0, name='李四', age=20, salary=6000.0}
  80.     //User{id=0, name='张三', age=50, salary=8000.0}
  81.     //User{id=0, name='编程指南', age=30, salary=120000.0}
  82. }
  84. //迭代 forEach
  85. @Test
  86. public void foreach() {
  87.     stream.forEach(System.out::println);
  88.     //输出
  89.     //User{id=0, name='张三', age=18, salary=6000.0}
  90.     //User{id=0, name='张三', age=18, salary=6000.0}
  91.     //User{id=0, name='张三', age=18, salary=6000.0}
  92.     //User{id=0, name='李四', age=20, salary=6000.0}
  93.     //User{id=0, name='张三', age=50, salary=8000.0}
  94.     //User{id=0, name='编程指南', age=30, salary=120000.0}
  96.     //stream 会帮我们自动实现内部迭代  forEach
  97.     //区别于外部迭代
  98.     list.forEach(System.out::println);
  99.     //其本质就是以下代码
  100.     Iterator<User> iterator = list.iterator();
  101.     while (iterator.hasNext()) {
  102.         System.out.println(iterator.next());
  103.     }
  104.     //也就是
  105.     for (User user : list) {
  106.         System.out.println(user);
  107.     }
  109. }
  111. //过滤 filter
  112. @Test
  113. public void test() {
  114.     Stream<User> userStream = stream.filter(user -> {
  115.         System.out.println("判断年龄……");
  116.         return user.getAge() > 18;
  117.     });
  118.     //此时运行无任何输出
  120.     //调用终止操作
  121.     userStream.forEach(System.out::println);
  123.     //输出如下：
  124.     //判断年龄……
  125.     //判断年龄……
  126.     //判断年龄……
  127.     //判断年龄……
  128.     //User{id=0, name='李四', age=20, salary=6000.0}
  129.     //判断年龄……
  130.     //User{id=0, name='张三', age=50, salary=8000.0}
  131.     //判断年龄……
  132.     //User{id=0, name='编程指南', age=30, salary=120000.0}
  135.     //说明： 中间操作不会自动执行，只有调用终止操作时，才会一次性执行  这个叫 惰性求值
  136. }
  138. //截断 limit
  139. @Test
  140. public void limit() {
  141.     stream.limit(3).forEach(System.out::println);
  143.     //输出如下：
  144.     //User{id=0, name='张三', age=18, salary=6000.0}
  145.     //User{id=0, name='张三', age=18, salary=6000.0}
  146.     //User{id=0, name='张三', age=18, salary=6000.0}
  147. }
  149. //跳过 skip
  150. @Test
  151. public void skip() {
  152.     stream.skip(3).forEach(System.out::println);
  154.     //输出如下：
  155.     //User{id=0, name='李四', age=20, salary=6000.0}
  156.     //User{id=0, name='张三', age=50, salary=8000.0}
  157.     //User{id=0, name='编程指南', age=30, salary=120000.0}
  158. }
  160. //去重 distinct
  161. @Test
  162. public void distinct() {
  163.     //distinct 需要重写 equals 和 hasCode方法
  164.     stream.distinct().forEach(System.out::println);
  166.     //输出如下：
  167.     //User{id=0, name='张三', age=18, salary=6000.0}
  168.     //User{id=0, name='李四', age=20, salary=6000.0}
  169.     //User{id=0, name='张三', age=50, salary=8000.0}
  170.     //User{id=0, name='编程指南', age=30, salary=120000.0}
  171. }
  173. //映射 map
  174. @Test
  175. public void map() {
  176.     //假如要提取用户名
  177.     stream.map(User::getName)
  178.         .forEach(System.out::println);
  179.     //map方法接收一个Function函数：接收一个值，处理后返回一个新值
  180.     //就是把每个对象映射成一个新对象
  181.     //  <R> Stream<R> map(Function<? super T, ? extends R> mapper);
  182.     //@FunctionalInterface
  183.     //public interface Function<T, R> {
  184.     //    R apply(T t);
  185.     //}
  187.     //输出如下：
  188.     //张三
  189.     //张三
  190.     //张三
  191.     //李四
  192.     //张三
  193.     //编程指南
  194. }
  196. //再来个映射测试  问每个人的薪水是多少
  197. @Test
  198. public void map1() {
  199.     stream.map(user -> user.getName() + "的薪水是： " + user.getSalary())
  200.         .forEach(System.out::println);
  201.     //map方法接收一个Function函数：接收一个值，处理后返回一个新值
  202.     //就是把每个对象映射成一个新对象
  203.     //  <R> Stream<R> map(Function<? super T, ? extends R> mapper);
  204.     //@FunctionalInterface
  205.     //public interface Function<T, R> {
  206.     //    R apply(T t);
  207.     //}
  209.     //输出如下：
  210.     //张三的薪水是： 6000.0
  211.     //张三的薪水是： 6000.0
  212.     //张三的薪水是： 6000.0
  213.     //李四的薪水是： 6000.0
  214.     //张三的薪水是： 8000.0
  215.     //编程指南的薪水是： 120000.0
  216. }
  218. //扁平化的映射 flatMap
  219. //就是把每一个值映射成一个流，然后把所有的流连在一起变成一个流
  220. @Test
  221. public void flatMap() {
  222.     stream.flatMap(user -> getStream(user.getName()))
  223.         .forEach(System.out::println);
  224.     //<R> Stream<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper);
  225.     //输出：
  226.     //张
  227.     //三
  228.     //张
  229.     //三
  230.     //张
  231.     //三
  232.     //李
  233.     //四
  234.     //张
  235.     //三
  236.     //编
  237.     //程
  238.     //指
  239.     //南
  241.     // map 和flatMap 的区别
  242.     // 等同于 add和addAll的区别
  243.     // 例如 map ={aaaa,bbbb,cccc}
  244.     // 那么flatMap = {a,a,a,a,b,b,b,b,c,c,c,c}
  245.     //就是把多个流合并为一个流了， 联想list.addAll() 把多个list合并成一个list
  248. }
  250. private Stream<String> getStream(String str) {
  251.     return Arrays.stream(str.split(""));
  252. }
  254. //排序：自然排序 sorted() 就是按照字典排序
  255. @Test
  256. public void sorted() {
  257.     Stream.of(1, 4, 5, 5, 6, 2, 3).sorted()
  258.         .forEach(System.out::print);
  259.     //输出: 1234556
  260. }
  262. //排序：定制排序 sorted(Comparator<? super T> comparator) 就是定制 Comparator接口
  263. @Test
  264. public void sorted1() {
  265.     Stream.of(1, 4, 5, 5, 6, 2, 3)
  266.         //倒序排列
  267.         .sorted((x, y) -> y > x ? 1 : -1)
  268.         .forEach(System.out::print);
  269.     //输出: 1234556
  270. }
  272. //排序例子： 先按薪资排，然后按年龄排
  273. @Test
  274. public void sorted2() {
  275.     long now = System.currentTimeMillis();
  276.     list.stream().sorted((o1, o2) -> {
  277.         if (o1.getAge().equals(o2.getAge())) {
  278.             return o1.getSalary().compareTo(o2.getSalary());
  279.         } else {
  280.             return o1.getAge().compareTo(o2.getAge());
  281.         }
  282.     }).close();
  283.     System.out.println(System.currentTimeMillis() - now);
  284.     //速度测试： 百万数据
  285.     // parallelStream()是 37毫秒 51 36
  286.     //stream() 是31毫秒 41 35
  288.     //速度测试： 千万数据
  289.     // parallelStream()是 32毫秒
  290.     //stream() 是33毫秒
  292. }
  294. //对比一下速度
  295. @Test
  296. public void sorted3() {
  297.     long now2 = System.currentTimeMillis();
  298.     list.sort((o1, o2) -> {
  299.         if (o1.getAge().equals(o2.getAge())) {
  300.             return o1.getSalary().compareTo(o2.getSalary());
  301.         } else {
  302.             return o1.getAge().compareTo(o2.getAge());
  303.         }
  304.     });
  305.     //        for (User user : list) {
  306.     //            System.out.println(user);
  307.     //        }
  308.     System.out.println(System.currentTimeMillis() - now2);
  309.     //速度测试： 百万数据 67毫秒 73 61
  310.     //千万数据： 192毫秒
  312. }

三、结束操作

  1. //是否都匹配 allMatch
  2. boolean allMatch = list.stream().allMatch(user -> user.getAge().equals(18));
  3. System.out.println(allMatch);//输出false 不是所有用户都18岁
  5. //至少有一个匹配 anyMatch
  6. boolean anyMatch = list.stream().anyMatch(user -> user.getAge() == 18);
  7. System.out.println(anyMatch);//输出true  有18岁的用户
  9. //都不匹配 noneMatch
  10. boolean noneMatch = list.stream().noneMatch(user -> user.getAge() == 18);
  11. System.out.println(noneMatch);//输出false 不是所有用户都不是18岁
  13. //查找第一个 findFirst
  14. Optional<User> first = list.stream()
  15.     .sorted(Comparator.comparingDouble(User::getSalary))
  16.     .findFirst();
  17. System.out.println(first);
  18. //Optional 可选的，用来封装对象的容器 目的是减少空指针 我们后边会细说
  21. //查找任意一个
  22. Optional<User> any = list.parallelStream()
  23.     .filter(user -> user.getAge() > 18)
  24.     .findAny();
  25. System.out.println(any.get());
  27. //查找最大值
  28. Optional<User> max = list.stream().max(Comparator.comparingDouble(User::getAge));
  29. System.out.println(max);
  30. // 查年龄最大的用户 Optional[User{id=null, name='张三', age=50, salary=8000.0}]
  32. //查最小的
  33. //查最少的工资是多少
  34. Optional<Double> min = list.stream().map(User::getSalary)
  35.     .min(Double::compareTo);
  36. System.out.println(min);//输出6000
  38. //归约
  39. Stream<Integer> stream = Stream.of(1, 2, 3, 4, 5, 6,7,8,9,10);
  40. Integer reduce = stream.reduce(0, Integer::sum);
  41. System.out.println(reduce); //55
  42. //   Optional<T> reduce(BinaryOperator<T> accumulator);
  44. //归约 计算工资的总和
  45. Optional<Double> reduce1 = list.stream().map(User::getSalary)
  46.     .reduce(Double::sum);
  47. System.out.println(reduce1.get());//152000.0
  50. //收集
  51. list.stream()
  52.     .map(User::getName)
  53.     .collect(Collectors.toList())
  54.     .forEach(System.out::println);
  55. //  <R> R collect(Supplier<R> supplier,
  56. //                  BiConsumer<R, ? super T> accumulator,
  57. //                  BiConsumer<R, R> combiner);
  59. //输出：
  60. //张三
  61. //张三
  62. //张三
  63. //李四
  64. //张三
  65. //编程指南
  67. System.out.println("----------------");
  69. list.stream()
  70.     .map(User::getName)
  71.     .collect(Collectors.toSet())
  72.     .forEach(System.out::println);
  73. //输出：
  74. //李四
  75. //张三
  76. //编程指南
  77. System.out.println("==============");
  79. LinkedHashSet<String> linkedHashSet = list.stream()
  80.     .map(User::getName)
  81.     .collect(Collectors.toCollection(LinkedHashSet::new));
  82. System.out.println(linkedHashSet);
  83. //输出： [张三, 李四, 编程指南]
  85. //计算平均值
  86. Long count = list.stream()
  87.     .collect(Collectors.counting());
  88. //等同于
  89. //        Long collect = list.stream().count();
  90. System.out.println(count);
  91. //计算工资的平均值
  92. Double avg = list.stream().collect(Collectors.averagingDouble(User::getSalary));
  93. System.out.println(avg);
  94. //激素那年龄总和
  95. Double sum = list.stream().collect(Collectors.summingDouble(User::getAge));
  96. //等同于
  97. //        Double sum = list.stream().mapToDouble(User::getAge).sum(); 
  98. System.out.println(sum);
  100. //以上输出：
  101. //6
  102. //25333.333333333332
  103. //154.0
  106. System.out.println("-------分组----------");
  107. //分组
  108. //安年龄分组
  109. Map<Integer, List<User>> map = list.stream()
  110.     .collect(Collectors.groupingBy(User::getAge));
  111. System.out.println(map);
  112. //{50=[User{id=null, name='张三', age=50, salary=8000.0}],
  113. // 18=[User{id=null, name='张三', age=18, salary=6000.0},
  114. //   User{id=null, name='张三', age=18, salary=6000.0},
  115. //   User{id=null, name='张三', age=18, salary=6000.0}],
  116. // 20=[User{id=null, name='李四', age=20, salary=6000.0}],
  117. // 30=[User{id=null, name='编程指南', age=30, salary=120000.0}]}
  119. //多级分组
  120. System.out.println("-------------多级分组-----------");
  121. Map<String, Map<String, List<User>>> map2 = list.stream()
  122.     .collect(Collectors.groupingBy(User::getName,
  123.                                    Collectors.groupingBy(user -> {
  124.                     if (user.getAge() <= 18) {
  125.                         return "青少年";
  126.                     } else if (user.getAge() <= 60) {
  127.                         return "中年";
  128.                     } else {
  129.                         return "老年";
  130.                     }
  131. })));
  133. System.out.println(map2);
  134. //输出：{李四={中年=[User{id=null, name='李四', age=20, salary=6000.0}]},
  135. // 张三={青少年=[User{id=null, name='张三', age=18, salary=6000.0},
  136. //   User{id=null, name='张三', age=18, salary=6000.0},
  137. //   User{id=null, name='张三', age=18, salary=6000.0}],
  138. // 中年=[User{id=null, name='张三', age=50, salary=8000.0}]},
  139. // 编程指南={中年=[User{id=null, name='编程指南', age=30, salary=120000.0}]}}
  141. //收集统计
  142. System.out.println("-----收集统计-----");
  143. LongSummaryStatistics statistics = list.stream()
  144.     .collect(Collectors.summarizingLong(User::getAge));
  145. double average = statistics.getAverage();
  146. long count1 = statistics.getCount();
  147. long max1 = statistics.getMax();
  148. long min1 = statistics.getMin();
  149. long sum1 = statistics.getSum();
  151. //收集 连接
  152. String join = list.stream().map(User::getName).collect(Collectors.joining());
  153. System.out.println(join);
  154. //张三张三张三李四张三编程指南
  155. String join1 = list.stream().map(User::getName).collect(Collectors.joining(" , "));
  156. System.out.println(join1);
  157. //张三 , 张三 , 张三 , 李四 , 张三 , 编程指南

并行流和顺序流

  • 并行流 就是把一个内容分成多个数据块，并用不同的线程分别处理每个数据块的流。
  • Java 8 中将并行进行了优化，我们可以很容易的对数据进行并行操作。
  • Stream API 可以声明性地通过 parallel() 与sequential() 在并行流与顺序流之间进行切换。
  • 并行流底层使用的是JDK7的Fork/Join框架实现的，相当于提供了一个种简便的使用方式。

了解 Fork/Join 框架

Fork/Join 框架：就是在必要的情况下，将一个大任务，进行拆分(fork)成若干个小任务（拆到不可再拆时），再将一个个的小任务运算的结果进行 join 汇总.
image.png

Fork/Join框架相对于多线程执行的优势是，可以更高效的利用cpu，采用 “工作窃取”模式（work-stealing）：
当执行新的任务时它可以将其拆分分成更小的任务执行，并将小任务加到线程队列中，然后再从一个随机线程的队列中偷一个并把它放在自己的队列中。

相对于一般的线程池实现,fork/join框架的优势体现在对其中包含的任务的处理方式上.在一般的线程池中,如果一个线程正在执行的任务由于某些原因无法继续运行,那么该线程会处于等待状态.而在fork/join框架实现中,如果某个子问题由于等待另外一个子问题的完成而无法继续运行.那么处理该子问题的线程会主动寻找其他尚未运行的子问题来执行.这种方式减少了线程的等待时间,提高了性能 。

举个例子： 计算0到一千亿数字求和
**
一、传统循环累积求和法

  1. //for循环累加 
  2. @Test
  3. public void test2() {
  4.     long start = System.currentTimeMillis();
  5.     long sum = 0L;
  6.     for (long i = 0L; i <= 10000000000L; i++) {
  7.         sum += i;
  8.     }
  9.     System.out.println(sum);
  10.     long end = System.currentTimeMillis();
  11.     System.out.println("耗费的时间为: " + (end - start));
  12.     //耗费的时间为: 2660
  13. }

二、ForkJoin框架计算

  1. package com.initit.java8.streamapi;
  3. import java.util.concurrent.RecursiveTask;
  5. /**
  6.  * ForkJoin框架使用案例
  7.  * 计算求和
  8.  */
  9. public class ForkJoinCalculate extends RecursiveTask<Long> {
  10.     private long start;
  11.     private long end;
  13.     private static final long THRESHOLD = 10000L; //临界值
  15.     public ForkJoinCalculate(long start, long end) {
  16.         this.start = start;
  17.         this.end = end;
  18.     }
  20.     /**
  21.      * 小于临界值10000的时候，累加求和
  22.      * 大于临界值10000的时候，对半拆分任务，交给多个线程执行
  23.      */
  24.     @Override
  25.     protected Long compute() {
  26.         long length = end - start;
  27.         if (length <= THRESHOLD) {
  28.             long sum = 0;
  29.             for (long i = start; i <= end; i++) {
  30.                 sum += i;
  31.             }
  32.             return sum;
  33.         } else {
  34.             long middle = (start + end) / 2;
  35.             ForkJoinCalculate left = new ForkJoinCalculate(start, middle);
  36.             left.fork(); //拆分，并将该子任务压入线程队列
  37.             ForkJoinCalculate right = new ForkJoinCalculate(middle + 1, end);
  38.             right.fork();
  39.             return left.join() + right.join();
  40.         }
  41.     }
  43. }
  46. //测试
  47. @Test
  48. public void test1() {
  49.     long start = System.currentTimeMillis();
  50.     ForkJoinPool pool = new ForkJoinPool();
  51.     ForkJoinTask<Long> task = new ForkJoinCalculate(0L, 10000000000L);
  52.     long sum = pool.invoke(task);
  53.     System.out.println(sum);
  54.     long end = System.currentTimeMillis();
  55.     System.out.println("耗费的时间为: " + (end - start));
  56.     //耗费的时间为: 1145
  57. }

三、使用Stream API 并行流计算求和

  1.     @Test
  2.     public void test3() {
  3.         long start = System.currentTimeMillis();
  4.         Long sum = LongStream.rangeClosed(0L, 10000000000L)
  5.                 .parallel()  //切换为并行流
  6. //                .sequential() //切换为顺序流 串行
  7.                 .sum();
  8.         System.out.println(sum);
  9.         long end = System.currentTimeMillis();
  10.         System.out.println("耗费的时间为: " + (end - start));
  11.         //耗费的时间为: 755
  12.     }
  • .parallel() //切换为并行流
  • .sequential() //切换为顺序流 串行

map-reduce思想

最后简单提一下map-reduce思想。

  • map 映射，数学概念 把A映射成B，A和B有某种对应关系
  • reduce 归约，数学概念，把一系列数据按照某种关系累积

起源于google，用来处理大量的网页数据（大数据计算），后hadoop框架借鉴改思想。
map-reduce是一种并行处理数据的架构模式，就是利用大量的计算机来并行计算数据。

打个比喻：

  • We want to count all the books in the library. You count up shelf #1, I count up shelf #2. That’s map. The more people we get, the faster it goes.
  • Now we get together and add our individual counts. That’s reduce.
  • 我们要数图书馆中的所有书。你数1号书架，我数2号书架。这就是“Map”。我们人越多，数书就更快。
  • 现在我们到一起，把所有人的统计数加在一起。这就是“Reduce”。

知乎有位大神说：

  • 举个栗子，要斗地主了，需要从10副混一起的牌里找出一幅牌，10副牌分10堆给10个人去清，每个人分别把黑红梅方放一堆，这就是map
  • 然后安排4个人每人去清一种花色的牌，从中找出1到K，这就是reduce了

hadoop入门有一个统计单词出现次数的例子word count，我们简单分析一下，了解这种思想：

  1. MapReduce大体上分为六个步骤：input, split, map, shuffle, reduce, output。细节描述如下：
  2. 1. 输入(input)：如给定一个文档，包含如下四行：
  3. • Hello Java
  4. • Hello C
  5. • Hello Java
  6. • Hello C++
  7. 2. 拆分(split)：将上述文档中每一行的内容转换为key-value对，即：
  8. • 0 - Hello Java
  9. • 1 - Hello C
  10. • 2 – Hello Java
  11. • 3 - Hello C++
  12. 3. 映射(map)：将拆分之后的内容转换成新的key-value对，即：
  13. • (Hello , 1)
  14. • (Java , 1)
  15. • (Hello , 1)
  16. • (C , 1)
  17. • (Hello , 1)
  18. • (Java , 1)
  19. • (Hello , 1)
  20. • (C++ , 1)
  21. 4. 派发(shuffle)：将key相同的扔到一起去，即：
  22. • (Hello , 1)
  23. • (Hello , 1)
  24. • (Hello , 1)
  25. • (Hello , 1)
  26. • (Java , 1)
  27. • (Java , 1)
  28. • (C , 1)
  29. • (C++ , 1)
  30. 注意：这一步需要移动数据，原来的数据可能在不同的datanode上，这一步过后，相同key的数据会被移动到同一台机器上。最终，它会返回一个list包含各种k-value对，即：
  31. • { Hello: 1,1,1,1}
  32. • {Java: 1,1}
  33. • {C: 1}
  34. • {C++: 1}
  35. 5. 缩减(reduce)：把同一个key的结果加在一起。如：
  36. • (Hello , 4)
  37. • (Java , 2)
  38. • (C , 1)
  39. • (C++,1)
  40. 6. 输出(output): 输出缩减之后的所有结果。

数据流处理-Stream API - 图3

就是一种拆分和合并的思想，目的是使用多个计算资源（多台计算、多进程、多线程等等）并行计算，以提升计算效率，节省计算时间。是一种很好的架构模式，大数据计算框架Hadoop等都是采用map-reduce思想。