carpenterlee/javalambdainternals

深入理解Java函数式编程和Streams API

Java Functional Programming Internals

Authors

Name	Weibo Id	Blog	Mail
李豪	@计算所的小鼠标	CarpenterLee	[email protected]

Introduction

本文受启发于JavaOne 2016关于Lambda表达式的相关主题演讲Lambdas and Functional Programming和Refactoring to Java 8。

Java 8已经发行两年多，但很多人仍然在使用JDK7。对企业来说，技术上谨慎未必是坏事，但对个人学习而言，不去学习新技术就很可能被技术抛弃。Java 8一个重要的变更是引入函数式编程和Lambda表达式(lambda expression)，这听起来似乎很牛，有种我虽然不知道Lambda表达式是什么，但我仍然觉得很厉害的感觉。

Java stream包是跟Lambda表达式同时添加新功能。Lambda表达式只有跟stream一起使用才能显示其真实的威力。

本系列文章不打算去争论“什么才算是一门真正的函数式语言”这类问题。我们会将重点放在如何使用Java Lambda表达式，如何使用stream，以及二者背后的原理。

Contents

具体内容安排如下：

1. Lambda and Anonymous Classes(I)，展示如何使用Lambda表达式替代匿名内部类，说明Lambda表达式和函数接口的关系。
2. Lambda and Anonymous Classes(II)，Lambda表达式的实现原理
3. Lambda and Collections，学习Java集合框架（Java Collections Framework）新加入的方法
4. Streams API(I)，Stream API基本用法
5. Streams API(II)，Stream规约操作用法，顺道说明接口静态方法和默认方法以及方法引用的概念。
6. Stream Pipelines，Stream流水线的实现原理
7. ParallelStream，Stream并行实现原理。
8. Stream Performance，Stream API性能评测。

javalambdainternals's Issues

关于sink设计的一个小疑问

作者大大你好，我似乎陷入了某种困境，在读完上面这句话：你可能会觉得这很简单，只需要从流水线的head开始依次执行每一步的操作（包括回调函数）就行了。这听起来似乎是可行的，但是你忽略了前面的Stage并不知道后面Stage到底执行了哪种操作，以及回调函数是哪种形式。换句话说，只有当前Stage本身才知道该如何执行自己包含的动作。这就需要有某种协议来协调相邻Stage之间的调用关系。

我竟然感觉"只需要从流水线的head开始依次执行每一步的操作（包括回调函数）就行了"这句话没问题。。。
1.当前stage做好自己事就好了吧，为啥要知道后面的stage呢？
2.尽管使用了sink，也还是当前sink做自己事，也没有管后面的sink额。。
3.如果说是为了协调调用关系的话，stage本来也就是双链的，上一个是谁，下一个是谁，清楚的很。
哭了，所以我不太清楚设计sink的目的，看起来似乎只是把处理数据的操作单独拎了出来。。。求解惑呀../(ㄒoㄒ)/~~

疑问

Map<Department, Integer> totalByDept = employees.stream()
.collect(Collectors.groupingBy(Employee::getDepartment,
Collectors.counting()));// 下游收集器
Collectors.counting()怎么返回Integer

在使用steam并行的计算时候，出现了oom

感谢您的科普，我也顺利用上了流。
在实际使用并行流的时候遇到了一个oom。
大致情况是要处理一个四十多m的txt文件，内容为换行的字符串。
使用lines.flatMap(line -> Arrays.stream(line.split(" "))).parallel().distinct().count();做并发计算词语个数（数量大概在三百万，类似
一@一对一 5
一@一道 5
一@丁 6
一@七旬 8
一@万 157）。
执行期间，并行流吃光了我的内存，cpu，我jvm中的old跟eden全部爆满，方法区正常，线程正常。
串行与外部迭代只会吃掉五十m内存。

请教作者，是我并发流用的不对，还是jdk8的流有bug ---下面是重现代码
static void useStream(){
long uniqueWord = 0;
try (Stream lines =
Files.lines(Paths.get("D:\nlp\dictionary\CoreNatureDictionary.ngram.txt"), Charset.defaultCharset())){
long start = System.currentTimeMillis();
uniqueWord = lines.flatMap(line -> Arrays.stream(line.split(" "))).parallel().distinct().count();
// uniqueWord = lines.flatMap(line -> Arrays.stream(line.split(" "))).distinct().count();
System.out.println(uniqueWord);
long end = System.currentTimeMillis();
System.out.println("耗时"+ (end-start)+"ms");
}catch (IOException e){
e.printStackTrace();
}
}

疑问

Stream<String> stream = Stream.of("I", "love", "you", "too");
List<String> list = stream.collect(ArrayList::new,ArrayList::add,(t,u)-> {
			System.out.println("t:" + t+" u:" + u);
			t.addAll(u);
});
System.out.println(list);

谢谢, 你启发(正反参半)了我.

这个例子,collect的第三个参数并没有执行,若能举一个体现第三个参数作用的例子会更好.

第7篇什么时候出

请教下sink之间是倒序查找的，那么如何保证执行操作的正确性？

举个例子：

String[] s = {"1.0", "2.0", "3.0", "4.0", "5.0"};
Double ans = Stream.of(s)
		.map(Double::valueOf)
		.filter(d -> d > 2)
		.sorted(Comparator.reverseOrder())
		.limit(2)
		.reduce(Double::sum)
		.orElse(0d);

倒序执行第一个Intermediate Operation 时，也就是limit，如何保证不会获取"1.0", "2.0"？