[Java知识库]函数式编程——Java Stream API

前言

1、lambda

lambda表达式又称闭包或匿名函数，可用于简化代码、增强代码可读性、并行操作集合等。

基本语法：

(parameters) -> expression
or
(parameters) ->{ statements; }

需注意的点：

lambda表达式内，不可以改变表达式外变量的值。

 List<User> userList = new ArrayList<>();
 int i=0;
 userList.forEach(k->{
     i=4;
    }
 );

Variable used in lambda expression should be final or effectively final

2、函数式接口

Functional Interface
一个有且仅有一个抽象方法，但是可以有多个非抽象方法的接口。

JDK 1.8 之前已有的函数式接口:

java.lang.Runnable
java.util.concurrent.Callable
java.security.PrivilegedAction
java.util.Comparator
java.io.FileFilter
java.nio.file.PathMatcher
java.lang.reflect.InvocationHandler
java.beans.PropertyChangeListener
java.awt.event.ActionListener
javax.swing.event.ChangeListener

JDK 1.8 新增加的函数接口：

java.util.function包下，如：
Function
Predicate
Supplier
Consumer

注意：
函数式接口不一定要使用@FunctionalInterface，加上该注解是为了方便编译器检查。

3、函数式编程

什么是函数？

举个例子

举个例子：求1~100之间数字相加之和。
命令式编程：

        long start=System.nanoTime();
        Integer sum=0;
        for (int i = 1; i < 101; i++) {
            sum=sum+i;
        }
        long end=System.nanoTime();
        System.out.println(end-start);   //49828

函数式编程：

		long start1=System.nanoTime();
        IntStream.range(1,101).sum();
        long end1=System.nanoTime();
        System.out.println(end1-start1); //61774697

比较可以看出，作为一种新的编程范式，函数式编程有着非常明显的优势：

● 简洁
● 无需关系内部实现细节，只要告知要做什么即可，具体如何做我们不需要关心。

函数式编程使用一系列的函数解决问题。函数仅接受输入并产生输出，不包含任何能影响产生输出的内部状态。任何情况下，使用相同的参数调用函数始终能产生同样的结果。

● 函数是一等公民
● 匿名函数lambda
● 封装控制结构的内置模板函数 （避免使用变量）
● 闭包

Java 8

Java8中的Stream是对容器对象功能的增强，它专注于对容器对象进行各种非常便利、高效的 聚合操作（aggregate operation），或者大批量数据操作 (bulk data operation)。Stream API借助于同样新出现的Lambda表达式，极大的提高编程效率和程序可读性。

Java从8开始引入stream API，虽然是同步流，但还是先来熟悉下基本操作吧。

步骤：

举个例子：

//准备数据
List<User> list=new ArrayList<>();
        for (int i=0;i<5;i++){
            User user=new User();
            user.setId(i);
            user.setName("name"+i);
            list.add(user);
        }
        
//示例一：
List<User> userList=list.stream()
  .filter(k-> k.getId()>0)
  .collect(Collectors.toList());
//打印结果：
//[User{id=1, name='name1'}, User{id=2, name='name2'}, User{id=3, name='name3'}, User{id=4, name='name4'}]


//示例二：
String names=list.stream().map(User::getName).collect(Collectors.joining(","));
//打印结果：
//name0,name1,name2,name3,name4

示例一是筛选出list中id>0的元素，生成新的集合userList。
其中：
stream()——将数据集合转换为流
filter()——中间操作，这里是根据自定义条件过滤元素。
collect()——终止操作，最终的数据流以何种方式展现。

同样的，示例二是取出User集合中的name列，并将其用逗号拼接在一起，最终返回一个字符串names。

符号::

选择器，上述写法等价于
String names=list.stream().map(k->k.getName()).collect(Collectors.joining(","));

总结一下，流处理主要分三个步骤，创建流、中间操作和最终操作。

1、创建流

创建的方式：
最常用的是从集合或数据中创建流。

 //Collection提供了接口的默认实现，创建的流有两种类型：并行流和非并行流。

default Stream<E> stream() {
     return StreamSupport.stream(this.spliterator(), false);
}

default Stream<E> parallelStream() {
     return StreamSupport.stream(this.spliterator(), true);
}

//Arrays提供的方法
public static <T> Stream<T> stream(T[] array) {
     return stream((Object[])array, 0, array.length);
 }

//Stream类提供的方法
static <T> Stream<T> of(T... values) {
     return Arrays.stream(values);
}

2、中间操作

中间操作的方法，返回的内容还是Stream，只是对Stream做了一些操作。

map (mapToInt, flatMap 等)、 filter、 distinct、 sorted、 peek、 limit、 skip、 parallel、 sequential、 unordered

3、终止操作

终止操作的方法，返回的内容不再是Stream，而是数据类型。

常见的终止操作方法：

forEach、 forEachOrdered、 toArray、 reduce、 collect、 min、 max、 count、 anyMatch、 allMatch、 noneMatch、 findFirst、 findAny、 iterator

操作：

filter（筛选）

使用场景：
需要根据筛选条件来产生一个子集合时。

map（映射）

使用场景：
对原集合中每一个元素执行给定的函数，得出一个新的集合。

int[] ints=IntStream.range(1,5).map(k->k*2).toArray();

reduce/fold（折叠/化约）

使用场景：
需要把集合分成一小块一小块来处理时。

//0,1,2,3,4
int asInt = list.stream().mapToInt(User::getId).reduce((x, y) -> x += y).getAsInt();
System.out.println(asInt);  //打印输出 10

用一个“累积量”来“收集”元素。

小结：

1、函数式的味道
之前：一个指令一个指令的发出，让计算机严格按照你的指令工作完成一个功能。
函数式：告知你要做什么，并不需要了解其具体如何运作。

控制权让渡语言/运行时：
使用高阶函数取代基本的控制结构，将细节交托给运行时。放弃对内存的直接控制。

改变少量的数据结构搭配大量的操作。

10个函数操作10中数据结构 => 100个函数操作一种数据结构

让语言去迎合问题，而非拿问题去硬套语言。

2、流只能使用一次

 		IntStream range = IntStream.range(1, 101);
        System.out.println(range.sum());
        System.out.println(range.sum());

“类似Iterator实例，是消耗品。用过之后必须重新生成新的stream才能再次操作。”

3、函数式编程不一定快
流式操作本身会耗费一定时间，如果数据量没有在百万级以上，函数式编程的效率优势并不明显。

4、记忆和缓求值/惰性

记忆：内部缓存
缓求值：尽可能推迟求解，必要时再执行。

5、闭包

6、柯里化和函数的部分施用

Java 9+

示例：

 public static void main(String[] args) {
        //创建发布者
        SubmissionPublisher<Integer> publisher=new SubmissionPublisher<>();
        //创建订阅者
        SimpleSubscriber subscriber1=new SimpleSubscriber("001",3);
        SimpleSubscriber subscriber2=new SimpleSubscriber("002",4);

        //订阅
        publisher.subscribe(subscriber1);
        publisher.subscribe(subscriber2);

        IntStream.range(1,5).forEach(item ->{
            publisher.submit(item);
            sleep(item);
                });
        publisher.close();
    }
    private static void sleep(Integer item){
        try {
            System.out.printf("推送数据：%d。休眠 3 秒。%n", item);
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

//订阅者

public class SimpleSubscriber  implements Flow.Subscriber<Integer> {
    private String name;
    private long maxCount;
    private long currentCount;

    private Flow.Subscription subscription;
    SimpleSubscriber(String name,long count){
        this.name=name;
        this.maxCount=count;
    }
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription=subscription;
        System.out.println("consumer:"+name+",maxCount:"+maxCount);
        subscription.request(maxCount);
    }

    @Override
    public void onNext(Integer integer) {
        currentCount++;
        System.out.println("consumer:"+name+" receive a message");
        if(currentCount>=maxCount){
            System.out.println("consumer:"+name+" count"+currentCount+" canceling");
            subscription.cancel();
        }
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("consumer:"+name+"occur"+throwable);
    }

    @Override
    public void onComplete() {
        System.out.println("consumer:"+name+" completed");
    }
}

执行结果：

推送数据：1。休眠 3 秒。
consumer:001,maxCount:3
consumer:002,maxCount:4
consumer:002 receive a message
consumer:001 receive a message
推送数据：2。休眠 3 秒。
consumer:002 receive a message
consumer:001 receive a message
推送数据：3。休眠 3 秒。
consumer:001 receive a message
consumer:002 receive a message
consumer:001 count3 canceling
推送数据：4。休眠 3 秒。
consumer:002 receive a message
consumer:002 count4 canceling

Process finished with exit code 0

源码分析

package java.util.concurrent;

public final class Flow {
    static final int DEFAULT_BUFFER_SIZE = 256;

    private Flow() {
    }

    public static int defaultBufferSize() {
        return 256;
    }

    public interface Processor<T, R> extends Flow.Subscriber<T>, Flow.Publisher<R> {
    }

    public interface Subscription {
        void request(long var1);

        void cancel();
    }

    public interface Subscriber<T> {
        void onSubscribe(Flow.Subscription var1);

        void onNext(T var1);

        void onError(Throwable var1);

        void onComplete();
    }

    @FunctionalInterface
    public interface Publisher<T> {
        void subscribe(Flow.Subscriber<? super T> var1);
    }
}

生产者

SubmissionPublisher

实践分享

1、在目前业务项目中，函数式编程模式并不会完全代替面向对象等命令编程。
无论是应用场景还是性能效率上，都不可能完全代替。但可以使用部分函数式风格的语言特性。

做好编码风格统一。

2、考虑是否需要在流处理中做额外处理
额外处理包括变量处理、远程调用等等。

3、考虑是否需要做异常处理、防御性检查

4、Optional对service层的侵入

附录：
《Pro Spring MVC with WebFlux Web Development in Spring Framework 5 and Spring Boot 2 by Marten Deinum Iuliana Cosmina (z-lib.org)》

《函数式编程思维》

代码示例：