了解Java8 Stream。

JDK8不久前正式发布了，这一次更新为我们带来了相当多的东西，其中一个非常庞大也是非常复杂的框架Stream，特别值得一提，他可以将一些让人烦躁的事情变得简洁，好，进入正题。

 

现在需要除去一个无序数组中大于800的数

 

传统方法：

		for(int i = 0; i < list.size(); i++) {
			if(list.get(i) > 800) {
				list.remove(i);
			}
		}

 使用Stream与lambda

list.stream().filter((n) -> n <= 800);

 

如此痛快已经显而易见了，但对于没有了解lambda表达式的童鞋们对类似代码有些抵触，接下来我们还使用传统的Java代码介绍几个比较常用的API。

	public static void learnMore(List<Integer> list) {
		// 创建流式调用对象
		list.stream()
		
		// 过滤，只保留值大于800的元素。
		.filter(new Predicate<Integer>() {

			@Override
			public boolean test(Integer t) {
				return t > 800;
			}
			
		})
		
		// 排序，默认升序，可传入Comparator。
		.sorted()
		
		// 通过map方法将集合中的对象转换为String类型
		.map(new Function<Integer, String>() {
			@Override
			public String apply(Integer t) {
				return new StringBuilder("String[").append(t).append("]").toString();
			}
		})
		
		// 获取集合中最多n个元素。
		.limit(10)
		
		.forEach(new Consumer<String>() {
			@Override
			public void accept(String t) {
				System.out.println(t);
			}
		});
		
		System.out.println("List size:" + list.size());
	}

 程序输出：

String[801]

String[802]

String[803]

String[804]

String[805]

String[806]

String[807]

String[808]

String[809]

String[810]

List size:1000

通过输出我们可以看到整个流程中所有的操作得到了处理，且所有操作均没有涉及对原集合的变动，探查源代码发现stream对象是靠原集合创建迭代器进行所有操作，因此不对原集合内容进行改动。

 

接下来我们看一个比较有意思的事情：

	public static void whyNoPrint(List<Integer> list) {
		list.stream()
		
		.filter(new Predicate<Integer>() {

			@Override
			public boolean test(Integer t) {
				System.out.println(t);
				return true;
			}
		});
	}

该方法执行过后程序没有任何输出，其实本人最初并不知道这个细节，而是在研究另外一个事情的时候发现的，稍微想一想就能看出这个程序有问题，流对象在filter后就不做任何事情了，所以这个筛选操作也失去了意义，应该是JDK在设计时考虑到的细节，非常值得我们学习。

 

接下来的东西实在是让人太兴奋了：并行流

这个东西是在研究stream.reduce方法时候发现的，reduce提供三个方法的重载，分别为1个参数，两个参数与三个参数，在尝试使用三个参数的reduce方法时候我发现了一个问题，就是最后一个参数传入的BinaryOperator.apply方法根本没有被调用！通过一番苦苦的读源代码工作后我发现reduce方法的第三个参数仅仅在stream以并行模式工作时才会生效！为什么？他是做什么的？我们来看下面的代码：

	public static void learnReduce_3(List<Integer> list) {
		// 并行流
		Stream<Integer> parallelStream = list.stream().parallel();
		// 串行流
		Stream<Integer> sequentialStream = list.stream();
		
		long start = System.currentTimeMillis();
		parallelStream.reduce(0, new BiFunction<Integer, Integer, Integer>() {

			@Override
			public Integer apply(Integer t, Integer u) {
				// 在这里停顿10毫秒来提高并行与串行计算的差异。
				// 虽然流提供了并行计算支持，但如果你需要处理的问题非常简单，还是推荐使用串行模式。
				// 因为那样避免了并行计算自身的开销，会更快。
				try {
					Thread.sleep(10);
				} catch (InterruptedException e) {
				}
				return t + u;
			}
		}, new BinaryOperator<Integer>() {

			@Override
			public Integer apply(Integer left, Integer right) {
				System.out.println(Thread.currentThread().getName());
				return left + right;
			}
		});
		System.out.println("Parallel stream processed with " + (System.currentTimeMillis() - start) + "ms.");
		
		start = System.currentTimeMillis();
		sequentialStream.reduce(0, new BiFunction<Integer, Integer, Integer>() {

			@Override
			public Integer apply(Integer t, Integer u) {
				try {
					Thread.sleep(10);
				} catch (InterruptedException e) {
				}
				return t + u;
			}
		}, new BinaryOperator<Integer>() {

			@Override
			public Integer apply(Integer t, Integer u) {
				System.out.println("Can't print this line.");
				return 0;
			}
		});
		System.out.println("Sequential stream processed with " + (System.currentTimeMillis() - start) + "ms.");
	}

 程序输出：

ForkJoinPool.commonPool-worker-3

ForkJoinPool.commonPool-worker-7

ForkJoinPool.commonPool-worker-2

main

ForkJoinPool.commonPool-worker-6

ForkJoinPool.commonPool-worker-4

ForkJoinPool.commonPool-worker-1

ForkJoinPool.commonPool-worker-1

ForkJoinPool.commonPool-worker-5

ForkJoinPool.commonPool-worker-3

ForkJoinPool.commonPool-worker-3

ForkJoinPool.commonPool-worker-7

ForkJoinPool.commonPool-worker-7

ForkJoinPool.commonPool-worker-2

ForkJoinPool.commonPool-worker-2

ForkJoinPool.commonPool-worker-2

ForkJoinPool.commonPool-worker-5

ForkJoinPool.commonPool-worker-5

ForkJoinPool.commonPool-worker-5

ForkJoinPool.commonPool-worker-5

ForkJoinPool.commonPool-worker-4

ForkJoinPool.commonPool-worker-4

ForkJoinPool.commonPool-worker-4

main

main

ForkJoinPool.commonPool-worker-6

ForkJoinPool.commonPool-worker-7

ForkJoinPool.commonPool-worker-7

ForkJoinPool.commonPool-worker-7

ForkJoinPool.commonPool-worker-7

ForkJoinPool.commonPool-worker-7

Parallel stream processed with 1350ms.

 

Sequential stream processed with 10003ms.

 

通过线程名称可以看出，流以并行模式运行的时候使用到了ForkJoin框架。

那么就更加明确了，当流在并行模式下工作时，JDK使用了多个线程进行并行处理，这将牵扯到多个计算结果合并的问题，所以reduce方法的第三个参数（BinaryOperator.apply）作用是将两个线程的计算结果合并。

 

更多信息参考：http://www.infoq.com/cn/articles/forkjoin-to-parallel-streams

 

至此，Stream基本上都已经搞清楚了，但是我们公司还在使用JDK6，呵呵！