万字长文详解Java lambda表达式

简介: 详细介绍java lambda的各种使用方式以及lambda的实行原理和序列化原理

本文的脉络如下

image.png

Lambda介绍

何为lambda

咱们首先来说说 Lambda 这个名字,Lambda 并不是一个什么的缩写,它是希腊第十一个字母 λ 的读音,同时它也是微积分函数中的一个概念,所表达的意思是一个函数入参和出参定义,在编程语言中其实是借用了数学中的 λ,并且多了一点含义,在编程语言中功能代表它具体功能的叫法是匿名函数(Anonymous Function),根据百科的解释:

匿名函数(英语:Anonymous Function)在计算机编程中是指一类无需定义 标识符(函数名)的 函数子程序

接着再来说说Lambda 的历史,虽然它在 JDK8 发布之后才正式出现,但是在编程语言界,它是一个具有悠久历史的东西,最早在 1958 年在Lisp 语言中首先采用,而且虽然Java脱胎于C++,但是C++在2011年已经发布了Lambda 了,但是 JDK8 的 LTS 在2014年才发布,所以 Java 被人叫做老土不是没有原因的,现代编程语言则是全部一出生就自带 Lambda 支持,所以Lambda 其实是越来越火的一个节奏~

Lambda 在编程语言中往往是一个**匿名函数**,也就是说Lambda 是一个抽象概念,而编程语言提供了配套支持,比如在 Java 中其实为Lambda 进行配套的就是函数式接口,通过函数式接口生成匿名类和方法进行Lambda 式的处理。<br />那么,既然是这一套规则我们明白了,那么Lambda 所提供的好处在Java中就是函数式接口所提供的能力了,函数式接口往往则是提供了一些通用能力,这些函数式接口在JDK中也有一套完整的实践,那就是 **Stream**。


不同语言中的Lambda

Python

lambda [arg1[,arg2,arg3....argN]]:expression

例子:

add2 = lambda x,y:x+y
print add2(1,2)     #3

sum2 = lambda x,y=10:x+y
print sum2(1)       #11
print sum2(1,100)   #101

C++

C++11中增加了对lambda表达式的支持

[ capture clause ] (parameters) -> return-type  
{   
   definition of method   
} 

具体语法:
image.png
[1]:Lambda表达式的引入标志,在‘[]’里面可以填入‘=’或‘&’表示该lambda表达式“捕获”(lambda表达式在一定的scope可以访问的数据)的数据时以什么方式捕获的,‘&’表示一引用的方式;‘=’表明以值传递的方式捕获,除非专门指出。
[2]:Lambda表达式的参数列表
[3]:Mutable 标识
[4]:异常标识
[5]:返回值
[6]:“函数”体,也就是lambda表达式需要进行的实际操作。
例子:

void func(std::vector<int>& v) {
  std::for_each(v.begin(), v.end(), [](int i) {
      cout << i << endl;
  });
}

Javascript

(p1 [,p2,p3,....pn]) => { code block  }

例子:

let func  = x => x * x;
func(2) #4

Java Lambda 表达式

Lambda 表达式在 Java 8 中添加的。
lambda 表达式是一小段代码,它接受参数并返回一个值。Lambda 表达式类似于方法,但它们不需要名称,并且可以直接在方法体中实现。

句法

最简单的 lambda 表达式包含一个参数和一个表达式:
零参数:

() -> System.out.println("零参数 lambda");

一个参数:

p -> System.out.println("一个参数:" + p);

多个参数:

(p1 [,p2,p3,....pn]) -> System.out.println("多个参数:" + p1 + ", " + p2 + ... + pn);

上面的表达式有一定的限制。它们要么返回一个值要么执行一段方法,并且它们不能包含变量、赋值或语句,例如if or for 。为了进行更复杂的操作,可以使用带有花括号的代码块。如果 lambda 表达式需要返回一个值,那么代码块应该有一个return语句。

(parameter1, parameter2) -> { code block [return] }

方法引用

  • 类 :: 静态方法
Consumer<String> c = [ (s) -> System.out.println(s);  <=>  System.out::println; ]
  • 对象 :: 实例方法
List<String> list = Lists.newArrayList();
Consumer<String> c = [ (e) => list.add(e);  <=>  list::add; ] 
  • 构造器 :: new
Supplier<List<String>> s = [ () -> new ArrayList<>(); <=> ArrayList::new; ]

原生函数式接口

@FunctionalInterface注解

有且只有一个抽象方法的接口被称为函数式接口,函数式接口适用于函数式编程的场景,Lambda就是Java中函数式编程的体现,可以使用Lambda表达式创建一个函数式接口的对象,一定要确保接口中有且只有一个抽象方法,这样Lambda才能顺利的进行推导。
与@Override 注解的作用类似,Java 8中专门为函数式接口引入了一个新的注解:@FunctionalInterface 。该注解可用于一个接口的定义上,一旦使用该注解来定义接口,编译器将会强制检查该接口是否确实有且仅有一个抽象方法(equal和hashcode方法不算),否则将会报错。但是这个注解不是必须的,只要符合函数式接口的定义,那么这个接口就是函数式接口。

Consumer: 消费性接口

Consumer通过名字可以看出它是一个消费函数式接口,主要针对的是消费(1..n 入参, 无返回)这个场景,它的代码定义如下:

@FunctionalInterface
public interface Consumer<T> {
    void accept(T t);
}

通过泛型 T 定义了一个入参,但是没有返回值,它代表你可以针对这个入参做一些自定义逻辑,比较典型的例子是 forEach 方法。
例子:

List<String> list = Lists.newArrayList("1", "2", "3", "4", "5", "6");
list.foreach(System.out::println); //打印数组

Supplier: 供给型接口

Supplier通过名字比较难看出来它是一个场景的函数式接口,它主要针对的是说获取(无入参,有返回)这个场景,它的代码定义如下:

@FunctionalInterface
public interface Supplier<T> {
    T get();
}

通过泛型 T 定义了一个返回值类型,但是没有入参,它代表你可以针对调用方获取某个值,比较典型的例子是 Stream 中的 collect 方法,通过自定义传入我们想要取得的某种对象进行对象收集。
例子:

List<String> list = Lists.newArrayList("1", "2", "3", "4", "5", "6");
List<String> newList = list.stream().filter(x -> x >= 2).collect(Collectors.toList()); 
// 将大于等于2的数重新收集成一个集合,其中Collectors.toList()的函数原型为 
// new CollectorImpl<>((Supplier<List<T>>) ArrayList::new, List::add,(left, right) -> { left.addAll(right); return left; },CH_ID)
// 原型中的ArrayList::new即为Supplier类型

Function<T,R>: 函数型接口

Function 接口的名字不太能轻易看出来它的场景,它主要针对的则是 转换(有入参,有返回,其中T是入参,R是返回)这个场景,其实说转换可能也不太正确,它是一个覆盖范围比较广的场景,你也可以理解为扩展版的Consumer,接口定义如下:

@FunctionalInterface
public interface Function<T, R> {
    R apply(T t);
}

通过一个入参 T 进行自定义逻辑处理,最终得到一个出参 R,比较典型的例子是 Stream 中的 map 系列方法和 reduce 系列方法。
例子:

List<String> list = Lists.newArrayList("1", "2", "3", "4", "5", "6");
List<Integet> newList = list.stream().map(Integer::parseInt).collect(Collectors.toList());
// map将list中所有的元素的类型由 String 通过 Integer.parseInt的方式转换为Intger。 简单来说就是A => B;

Predicate: 断言型接口

Predicate主要针对的是判断(有入参,有返回,凡是返回的类型固定为Boolean。可以说Function 是包含Predicate的 )这个场景,它的代码定义如下:

@FunctionalInterface
public interface Predicate<T> {
    boolean test(T t);
}

通过泛型 T 定义了一个入参,返回了一个布尔值,它代表你可以传入一段判断逻辑的函数,比较典型的例子是 Stream 中的 filter方法。

List<String> list = Lists.newArrayList("1", "2", "3", "4", "5", "6");
List<String> newList = list.stream().filter(x -> x >= 2).collect(Collectors.toList()); 
// 将大于等于2的数重新收集成一个集合,filter中的 x -> x >= 2就是Predicate接口

Stream表达式

Stream,就是JDK8又依托于函数式编程特性为集合类库做的一个类库,它其实就是jdk提供的函数式接口的最佳实践。它能让我们通过lambda表达式更简明扼要的以流水线的方式去处理集合内的数据,可以很轻松的完成诸如:过滤、分组、收集、归约这类操作。
其中Stream的操作大致分为两类

  • 中间型操作
  • 终结型操作

其中转换型操作又分为有状态和无状态两类。有状态是本次的结果需要依赖于前面的处理结果,而无状态则是不依赖。简单来讲就是无状态方法可以互相调换位置,而有状态方法不能调换位置。

中间型操作

中间型操作就是返回值依旧是stream类型的方法。api如下:

API 功能说明 无状态操作
filter() 按照条件过滤符合要求的元素, 返回新的stream流。
map() 将已有元素转换为另一个对象类型,一对一逻辑,返回新的stream流
peek() 对stream流中的每个元素进行逐个遍历处理,返回处理后的stream流
flatMap() 将已有元素转换为另一个对象类型,一对多逻辑,即原来一个元素对象可能会转换为1个或者多个新类型的元素,返回新的stream流
limit() 仅保留集合前面指定个数的元素,返回新的stream流
skip() 跳过集合前面指定个数的元素,返回新的stream流
concat() 将两个流的数据合并起来为1个新的流,返回新的stream流
distinct() 对Stream中所有元素进行去重,返回新的stream流
sorted() 对stream中所有的元素按照指定规则进行排序,返回新的stream流
takeWhile() JDK9新增,传入一个断言参数当第一次断言为false时停止,返回前面断言为true的元素。
dropWhile() JDK9新增,传入一个断言参数当第一次断言为false时停止,删除前面断言为true的元素。

终结型操作

终结型操作与中间型相反,返回值是非Stream类型的。api如下:

API 功能说明
count() 返回stream处理后最终的元素个数
max() 返回stream处理后的元素最大值
min() 返回stream处理后的元素最小值
findFirst() 找到第一个符合条件的元素时则终止流处理
findAny() 找到任何一个符合条件的元素时则退出流处理,这个对于串行流时与findFirst相同,对于并行流时比较高效,任何分片中找到都会终止后续计算逻辑
anyMatch() 返回一个boolean值,类似于isContains(),用于判断是否有符合条件的元素
allMatch() 返回一个boolean值,用于判断是否所有元素都符合条件
noneMatch() 返回一个boolean值, 用于判断是否所有元素都不符合条件
*collect() 将流转换为指定的类型,通过Collectors进行指定
*reduce() 将一个Stream中的所有元素反复结合起来,得到一个结果
toArray() 将流转换为数组
iterator() 将流转换为Iterator对象
foreach() 无返回值,对元素进行逐个遍历,然后执行给定的处理逻辑

探究lambda运行的底层原理

源码分析

接下来通过一个例子Debug来探究下lambda运行的底层原理,实验代码如下:

Set<Integer> collect = list.stream()
                .filter(e -> e > 2)
                .sorted()
                .map(e -> e * 2)
                .collect(Collectors.toSet());
上诉例子可拆解成下面5部分:
  Stream<Integer> stream = list.stream();
  Stream<Integer> filterStream = stream.filter(e -> e > 2);
  Stream<Integer> sortedStream = filterStream.sorted();
  Stream<Integer> mapStream = sortedStream.map(e -> e * 2);
  Set<Integer> integers = mapStream.collect(Collectors.toSet());
  • list.stream()
@Override
    default Spliterator<E> spliterator() {
        return Spliterators.spliterator(this, 0);
    }

default Stream<E> stream() {
    return StreamSupport.stream(spliterator(), false);
}
public static <T> Stream<T> stream(Spliterator<T> spliterator, boolean parallel) {
    Objects.requireNonNull(spliterator);
    return new ReferencePipeline.Head<>(spliterator,
                                        StreamOpFlag.fromCharacteristics(spliterator),
                                        parallel);
}

list.stream()最终调用了ReferencePipeline.Head<>,返回一个Head对象。Head是ReferencePipeline的内部类。官方注释说此类是ReferencePipeline的源阶段。也是stream调用的起始阶段。
运行完这一方法返回ReferencePipeline.Head对象,对象的所有元素保存在sourceSpliterator中
image.png

  • stream.filter(e -> e > 2)

filter的方法原型如下:

public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
    Objects.requireNonNull(predicate);
    return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
                                         StreamOpFlag.NOT_SIZED) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
            return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
                @Override
                public void begin(long size) {
                    downstream.begin(-1);
                }

                @Override
                public void accept(P_OUT u) {
                    if (predicate.test(u))
                        downstream.accept(u);
                }
            };
        }
    };
}

StatelessOp是stream 无状态的基类,与之相对的是StatefulOp,stream有状态的基类。元素原型如下E_IN是上游元素的类型,E_OUT是当前阶段返回的类型。

abstract static class StatelessOp<E_IN, E_OUT>
            extends ReferencePipeline<E_IN, E_OUT> {
        StatelessOp(AbstractPipeline<?, E_IN, ?> upstream,
                    StreamShape inputShape,
                    int opFlags) {
            super(upstream, opFlags);
            assert upstream.getOutputShape() == inputShape;
        }

        @Override
        final boolean opIsStateful() {
            return false;
        }
    }

abstract static class StatefulOp<E_IN, E_OUT>
            extends ReferencePipeline<E_IN, E_OUT> {
        StatefulOp(AbstractPipeline<?, E_IN, ?> upstream,
                   StreamShape inputShape,
                   int opFlags) {
            super(upstream, opFlags);
            assert upstream.getOutputShape() == inputShape;
        }

        @Override
        final boolean opIsStateful() {
            return true;
        }
}

需要注意的是 filter等方法的构造方法 new StatelessOp<P_OUT, P_OUT>(this,StreamShape.REFERENCE,StreamOpFlag.NOT_SIZED) 会将this传入。StatulessOp的构造方法,会一直super到AbstractPipeline方法。注意到AbstractPipeline类的构造方法中打注释的地方。

AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
        if (previousStage.linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
    
        previousStage.linkedOrConsumed = true; // 
        previousStage.nextStage = this; // 注意打注释的语句
        this.previousStage = previousStage; //
    
        this.sourceOrOpFlags = opFlags & StreamOpFlag.OP_MASK;
        this.combinedFlags = StreamOpFlag.combineOpFlags(opFlags, previousStage.combinedFlags);
        this.sourceStage = previousStage.sourceStage;
        if (opIsStateful())
            sourceStage.sourceAnyStateful = true;
        this.depth = previousStage.depth + 1;
    }

简化就是双向链表加入节点的操作。

p.next = this;
this.pre = p;

运行完返回如下:注意看上一步对象的地址保存在当前对象的perviousStage中,而且当前对象增加predicate对象
image.png
如图所示:
image.png

  • filterStream.sorted()
@Override
    public final Stream<P_OUT> sorted() {
        return SortedOps.makeRef(this);
    }
OfRef(AbstractPipeline<?, T, ?> upstream) {
    super(upstream, StreamShape.REFERENCE,
        StreamOpFlag.IS_ORDERED | StreamOpFlag.IS_SORTED);
    this.isNaturalSort = true;
    // Will throw CCE when we try to sort if T is not Comparable
    @SuppressWarnings("unchecked")
        Comparator<? super T> comp = (Comparator<? super T>) Comparator.naturalOrder();
    this.comparator = comp;
}

和filter操作一样,将Sorted节点加入链表中同时设置标志位StreamOpFlag.IS_ORDERED|StreamOpFlag.IS_SORTED
image.png
运行完这一步结果如图
image.png

  • sortedStream.map(e -> e * 2)

map()方法跟filter()方法的执行逻辑很像,分析方法跟分析filter()方法一样

public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
    Objects.requireNonNull(mapper);
    return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
                                     StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
            return new Sink.ChainedReference<P_OUT, R>(sink) {
                @Override
                public void accept(P_OUT u) {
                    downstream.accept(mapper.apply(u));
                }
            };
        }
    };
}

不过与filter不同时的参数中增加了mapper参数,类型为function
image.png
执行完如图所示:
image.png

  • mapStream.collect(Collectors.toSet());

collect方法原型和Collectors.toSet()方法原型如下:

public final <R, A> R collect(Collector<? super P_OUT, A, R> collector) {
    A container;
    if (isParallel()
        && (collector.characteristics().contains(Collector.Characteristics.CONCURRENT))
        && (!isOrdered() || collector.characteristics().contains(Collector.Characteristics.UNORDERED))) {
        container = collector.supplier().get();
        BiConsumer<A, ? super P_OUT> accumulator = collector.accumulator();
        forEach(u -> accumulator.accept(container, u));
    }
    else {
        container = evaluate(ReduceOps.makeRef(collector));
    }
    return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)
        ? (R) container
        : collector.finisher().apply(container);
}
new CollectorImpl<>((Supplier<Set<T>>) HashSet::new, Set::add,
                    (left, right) -> {
                        if (left.size() < right.size()) {
                            right.addAll(left); return right;
                        } else {
                            left.addAll(right); return left;
                        }
                    },
                    CH_UNORDERED_ID)

在collect方法中会判断是否为并行流,不是的话会执行evaluate(ReduceOps.makeRef(collector)); ReduceOps.makeRef(collector)会返回类型为TerminalOp的参数,在evaluate方法中会将链表的节点都包装为Sink。


public static <T, I> TerminalOp<T, I>
    makeRef(Collector<? super T, I, ?> collector) {
    Supplier<I> supplier = Objects.requireNonNull(collector).supplier();
    BiConsumer<I, ? super T> accumulator = collector.accumulator();
    BinaryOperator<I> combiner = collector.combiner();
    class ReducingSink extends Box<I>
        implements AccumulatingSink<T, I, ReducingSink> {
            @Override
            public void begin(long size) {
                state = supplier.get();
            }

            @Override
            public void accept(T t) {
                accumulator.accept(state, t);
            }

            @Override
            public void combine(ReducingSink other) {
                state = combiner.apply(state, other.state);
            }
        }
    return new ReduceOp<T, I, ReducingSink>(StreamShape.REFERENCE) {
        @Override
        public ReducingSink makeSink() {
            return new ReducingSink();
        }

        @Override
        public int getOpFlags() {
            return collector.characteristics().contains(Collector.Characteristics.UNORDERED)
                ? StreamOpFlag.NOT_ORDERED
                : 0;
        }
    };
}
final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
    assert getOutputShape() == terminalOp.inputShape();
    if (linkedOrConsumed)
        throw new IllegalStateException(MSG_STREAM_LINKED);
    linkedOrConsumed = true;

    return isParallel()
        ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
        : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
}
public <P_IN> R evaluateSequential(PipelineHelper<T> helper,
                                   Spliterator<P_IN> spliterator) {
    return helper.wrapAndCopyInto(makeSink(), spliterator).get();
}
final <P_IN, S extends Sink<E_OUT>> S wrapAndCopyInto(S sink, Spliterator<P_IN> spliterator) {
    copyInto(wrapSink(Objects.requireNonNull(sink)), spliterator);
    return sink;
}
final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
    Objects.requireNonNull(wrappedSink);

    if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
        wrappedSink.begin(spliterator.getExactSizeIfKnown());
        spliterator.forEachRemaining(wrappedSink);
        wrappedSink.end();
    }
    else {
        copyIntoWithCancel(wrappedSink, spliterator);
    }
}

执行完如图:
image.png
关键在于上面的copyInfo方法,此方法是stream的启动方法。遍历元素调用第一节点的逻辑(filter)。然后在end方法中调用第二个节点的begin方法,begin方法又会调用第二个节点的逻辑,之后和第一个节点一样,调用end方法,触发第三个节点的begin方法..... 最后调用到最后一个节点将处理好的元素收集起来。
下面是最后一步map的节点的栈帧和运行数据和对应的方法。这一步结束后会将此次运行的stream元素都add到hashSet中。
downstream.accept - > Set::add;
mapper.apply -> e -> e * 2;
image.png
image.png
最后调用(Supplier)(helper.wrapAndCopyInto(makeSink(), spliterator)).get()方法将保存元素的容器获取出来。

并发流源码分析

修改代码增加并发流:

public static void main(String[] args) {
    List<Integer> list = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

    //        Set<Integer> collect = list.stream()
    //                .filter(e -> e > 2)
    //                .filter(e -> e < 8)
    //                .sorted()
    //                .map(e -> e * 2)
    //                .peek(System.out::println)
    //                .collect(Collectors.toSet());
    Stream<Integer> stream = list.stream().parallel(); // list.parallelStream()
    Stream<Integer> filterStream = stream.filter(e -> e > 2);
    Stream<Integer> sortedStream = filterStream.sorted();
    Stream<Integer> mapStream = sortedStream.map(e -> e * 2);
    Set<Integer> integers = mapStream.collect(Collectors.toSet());
    System.out.println(integers);
}

根据非并发流的分析直接来到最后一步collect。分歧在evaluate方法中,之前调用terminalOp.evaluateSequential,并发流则会调用terminalOp.evaluateParallel。
image.png

public <P_IN> R evaluateParallel(PipelineHelper<T> helper,
                                 Spliterator<P_IN> spliterator) {
    return new ReduceTask<>(this, helper, spliterator).invoke().get();
}

在evaluateParallel返回会执行ReduceTask累的构造方法,查看ReduceTask类发现继承AbstractTask类

private static final class ReduceTask<P_IN, P_OUT, R,
    S extends AccumulatingSink<P_OUT, R, S>>
    extends AbstractTask<P_IN, P_OUT, S, ReduceTask<P_IN, P_OUT, R, S>> {
        private final ReduceOp<P_OUT, R, S> op;

        ReduceTask(ReduceOp<P_OUT, R, S> op,
                   PipelineHelper<P_OUT> helper,
                   Spliterator<P_IN> spliterator) {
            super(helper, spliterator);
            this.op = op;
        }

        ReduceTask(ReduceTask<P_IN, P_OUT, R, S> parent,
                   Spliterator<P_IN> spliterator) {
            super(parent, spliterator);
            this.op = parent.op;
        }

        @Override
        protected ReduceTask<P_IN, P_OUT, R, S> makeChild(Spliterator<P_IN> spliterator) {
            return new ReduceTask<>(this, spliterator);
        }

        @Override
        protected S doLeaf() {
            return helper.wrapAndCopyInto(op.makeSink(), spliterator);
        }

        @Override
        public void onCompletion(CountedCompleter<?> caller) {
            if (!isLeaf()) {
                S leftResult = leftChild.getLocalResult();
                leftResult.combine(rightChild.getLocalResult());
                setLocalResult(leftResult);
            }
            // GC spliterator, left and right child
            super.onCompletion(caller);
        }
    }

继续往上查看

//AbstractTask extends CountedCompleter
abstract class AbstractTask<P_IN, P_OUT, R,
    K extends AbstractTask<P_IN, P_OUT, R, K>>
    extends CountedCompleter<R> {
    }
//AbstractTask extends ForkJoinTask
public abstract class CountedCompleter<T> extends ForkJoinTask<T> {}

通过idea工具可以更直观的查看继承关系,ReduceTask最终继承ForkJoinTask。 ForkJoinTask与ForkJoinPool线程有关系。
image.png
程序继续运行会调用new ReduceTask<>(this, helper, spliterator).invoke()****,invoke方法ForkJoinTask的启动方法。

public final V invoke() {
    int s;
    if (((s = doInvoke()) & ABNORMAL) != 0)
        reportException(s);
    return getRawResult();
}

private int doInvoke() {
    int s; Thread t; ForkJoinWorkerThread wt;
    return (s = doExec()) < 0 ? s :
        ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
        (wt = (ForkJoinWorkerThread)t).pool.
        awaitJoin(wt.workQueue, this, 0L) :
        externalAwaitDone();
}


final int doExec() {
    int s; boolean completed;
    if ((s = status) >= 0) {
        try {
            completed = exec();
        } catch (Throwable rex) {
            completed = false;
            s = setExceptionalCompletion(rex);
        }
        if (completed)
            s = setDone();
        }
    return s;
}

protected final boolean exec() {
    compute();
    return false;
}

最后跟着调用链回来到AbstractTask类中的compute方法

public void compute() {
    Spliterator<P_IN> rs = spliterator, ls; // right, left spliterators
    long sizeEstimate = rs.estimateSize();
    long sizeThreshold = getTargetSize(sizeEstimate);
    boolean forkRight = false;
    @SuppressWarnings("unchecked") K task = (K) this;
    while (sizeEstimate > sizeThreshold && (ls = rs.trySplit()) != null) {
        K leftChild, rightChild, taskToFork;
        task.leftChild  = leftChild = task.makeChild(ls);
        task.rightChild = rightChild = task.makeChild(rs);
        task.setPendingCount(1);
        if (forkRight) {
            forkRight = false;
            rs = ls;
            task = leftChild;
            taskToFork = rightChild;
        }
        else {
            forkRight = true;
            task = rightChild;
            taskToFork = leftChild;
        }
        taskToFork.fork();
        sizeEstimate = rs.estimateSize();
    }
    task.setLocalResult(task.doLeaf());
    task.tryComplete();
}

在调用taskToFork.fork()前查看下当前变量表:
image.png
taskToFork的具体内容如下: op属性是Collectors.toSet(),而之前对元素的处理方法都在helper字段中
image.png
执行fork会把所有当前的task(this)放在ForkJoinPool这个线程池中。

public final ForkJoinTask<V> fork() {
    Thread t;
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        ((ForkJoinWorkerThread)t).workQueue.push(this);
    else
        ForkJoinPool.common.externalPush(this);
    return this;
}

执行完查看Debug线程堆栈信息,看到除了main线程在运行外,同时还多了ForkJoinPool这个线程组
image.png
同时这个又会运行到之前调用过的doExec()方法中,以此形成将大任务分解成小任务的循环。
image.png
最后主线程再执行task.doLeaf()运行到定义好的方法lambda方法中,并将此结果收集到一起
image.png
最后简单介绍下ForkJoinPool的运行原理:
image.png
ForkJoinPool核心思想:分治
image.png
总结一下并行流的和串行流的区别:串行流是一个一个处理的,而并行流是把元素先分成n部分,然后给将这n部分放到一个线程池中执行,等各个线程把这n部分执行完毕后在将结果汇总起来在输出最终结果。并行流可以极大的加速stream的处理速度,不过需要注意的是,程序中的是各个并行流公用一个线程池。

JVM分析

先写一段简单的包含lambda的代码,编译后查看编译文件和字节码。

public static void main(String[] args) {
        Stream.of(1, 2, 3, 4, 5)
                .filter(x -> x > 2)
                .forEach(System.out::println);
    }

main方法字节码如下:

 0 iconst_5
 1 anewarray #2 <java/lang/Integer>
 4 dup
 5 iconst_0
 6 iconst_1
 7 invokestatic #3 <java/lang/Integer.valueOf : (I)Ljava/lang/Integer;>
10 aastore
11 dup
12 iconst_1
13 iconst_2
14 invokestatic #3 <java/lang/Integer.valueOf : (I)Ljava/lang/Integer;>
17 aastore
18 dup
19 iconst_2
20 iconst_3
21 invokestatic #3 <java/lang/Integer.valueOf : (I)Ljava/lang/Integer;>
24 aastore
25 dup
26 iconst_3
27 iconst_4
28 invokestatic #3 <java/lang/Integer.valueOf : (I)Ljava/lang/Integer;>
31 aastore
32 dup
33 iconst_4
34 iconst_5
35 invokestatic #3 <java/lang/Integer.valueOf : (I)Ljava/lang/Integer;>
38 aastore
39 invokestatic #4 <java/util/stream/Stream.of : ([Ljava/lang/Object;)Ljava/util/stream/Stream;>
42 invokedynamic #5 <test, BootstrapMethods #0>
47 invokeinterface #6 <java/util/stream/Stream.filter : (Ljava/util/function/Predicate;)Ljava/util/stream/Stream;> count 2
52 getstatic #7 <java/lang/System.out : Ljava/io/PrintStream;>
55 dup
56 invokestatic #8 <java/util/Objects.requireNonNull : (Ljava/lang/Object;)Ljava/lang/Object;>
59 pop
60 invokedynamic #9 <accept, BootstrapMethods #1>
65 invokeinterface #10 <java/util/stream/Stream.forEach : (Ljava/util/function/Consumer;)V> count 2
70 return

同时在生成的字节码中有一个名为lambda$main$0的方法,字节码如下:

0 aload_0
 1 invokevirtual #11 <java/lang/Integer.intValue : ()I>
 4 iconst_2
 5 if_icmple 12 (+7)
 8 iconst_1
 9 goto 13 (+4)
12 iconst_0
13 ireturn

先看main方法的字节码的第29行与第35行分别是invokedynamic #5 <test, BootstrapMethods #0> 和 invokedynamic #9 <accept, BootstrapMethods #1>
这两个字节码分别对应class文件的BootstrapMethods中。查看编译出来class文件的BootstrapMethods,有几个关键的地方,第一个是innerClasser。第二个是在BootstrapMethods出现java/lang/invoke/LambdaMetafactory.metafactory 以及对应的argument分别是#39, #40 和#41。通过描述符可知#39的入参为Object,返回为boolean,#40的入参为MethodHandle 具体的类型为com.yousheng.lambda.test.LambdaTest.lambda$main$0(Integer)boolean,#41的入参为Integer,返回为boolean
image.png
debug断点调试,会运行到BootstrapMethodInvoker的127,会执行MethodHandler的invokeExact方法,此方法是native方法。
image.png
最后通过jvm的解析转发调用会来到LambdaMetafactory的metafactory方法中
image.png
这方法的最后3个入参类型就是从class文件中看到那三个入参的类型。
image.png
同时jvm也通过调用者和方法名称以及方法描述符找到了最后需要调用的方法。
image.png
查看ImplMethod参数
image.png
member对象为就是#40具体含义为:com.yousheng.lambda.test.LambdaTest.lambda$main$0(Integer)boolean/invokeStatic。

  • com.yousheng.lambda.test.LambdaTest -- > 类名
  • lambda$main$0 -- >类中的方法名称
  • (Integer)boolean -- > 方法的描述符, (括号内的为入参类型,返回值为boolean)
  • invokeStatic --> 调用字节码。 在jvm中有5中invoke字节码指令,分别为
字节码类型 方法说明
invokestatic 用于调用静态方法
invokespecial 用于调用私有实例方法、构造器,以及使用 super 关键字调用父类的实例方法或构造器,和所实现接口的默认方法
invokevirtual 用于调用非私有实例方法
invokeinterface 用于调用接口方法
invokedynamic 用于调用动态方法(java7后增加)

继续查看栈帧发现此方法是由Jvm调用而来,metafactory的上一个方法是invokeStatic当时行号是-1所以说明是jvm内部方法
image.png

可以理一下整个流程。
首先jvm启动,运行方法, 发现字节码是中存在invokeDynamic,通过invokedynamic字节码对应的BootstrapMethods调用MethodHandles.lookup方法寻找调用类中与当前lambda对应的静态内部类方法,最后生成CallSite 调用点,最后调用真正的lambda方法。
在jvm启动的时候增加参数-Djdk.internal.lambda.dumpProxyClasses=可将生成出来的静态内部dump到指定的目录下

image.png
通过javap -v -p [生成的文件] 可查看相应字节码,下图就是生成的两个静态内部类
filter(x -> x > 2) 中的 x-> x > 2的字节码:
image.png
forEach(System.out::println) 中的 System.out::println 字节码:
image.png
最后因为System.out::println属于类 :: 静态方法的形式,所以在生成的字节码中存在“适配器”,即先将System.out通过静态方法赋值给对应的静态内部类,在通过调用lambda方法使用。
image.png
而正常的方法会直接调用Lambda中的lambda$main$0方法。
image.png

Lambda的序列化原理

lambda的本质在上面的探究中我们也能看到是静态内部类。序列化lambda跟序列化其他对象一样必须要实现Serializable接口, 为什么必须实现Serializable才能进行序列化呢,可以从源码中找到答案,ObjectOutputStream中的1178行。程序判断当前对象obj是否为Serializable子类,如果是的话进行序列化,否则抛出异常。注:如果需要实例化对象,那么这个对象里面的所有属性必须都是可实例化(即所有的属性包括自身都必须实现Serializable接口)。
image.png
知道序列化原理后,可以使用下列代码进行测试,并将运行时的class文件dump到本地。

 Function<Child, String> function = (Function<Child, String> & Serializable) Child::getName;

利用查看dump下的class文件,发现类实现Serializable接口,在类中又增加了writeReplace方法。且方法返回值为SerializedLambda。

final class LambdaTest$$Lambda$15 implements Function, Serializable {
    private LambdaTest$$Lambda$15() {
    }

    @Hidden
    public Object apply(Object var1) {
        return ((Child)var1).getName();
    }

    private final Object writeReplace() {
        return new SerializedLambda(LambdaTest.class, "java/util/function/Function", "apply", "(Ljava/lang/Object;)Ljava/lang/Object;", 5, "com/yousheng/lambda/entity/Child", "getName", "()Ljava/lang/String;", "(Lcom/yousheng/lambda/entity/Child;)Ljava/lang/String;", new Object[0]);
    }
}

先看一下writeReplace方法是什么,熟悉序列化的人知道如果类在进行序列化的时候会先查询类中是否有writeReplace方法。这一点同样可以在ObjectOutputStream类中1126行找到对应的处理逻辑。调用writeReplace方法返回的对象会替换原有的obj对象。
image.png
就是说实现了Serializable接口的lambda对象最后会被实例化成SerializedLambda类型,从SerializedLambda类上面的注释中可以看出来,类中的readResolve方法会去"capturing class"中寻找$deserializeLambda$(SerializedLambda)方法并会将"this"对象当做第一个参数传入。实现$deserializeLambda$的 Lambda 类负责验证SerializedLambda的属性是否与该类实际捕获的 lambda 一致。关键字:验证

/**
 * Serialized form of a lambda expression.  The properties of this class
 * represent the information that is present at the lambda factory site, including
 * static metafactory arguments such as the identity of the primary functional
 * interface method and the identity of the implementation method, as well as
 * dynamic metafactory arguments such as values captured from the lexical scope
 * at the time of lambda capture.
 *
 * <p>Implementors of serializable lambdas, such as compilers or language
 * runtime libraries, are expected to ensure that instances deserialize properly.
 * One means to do so is to ensure that the {@code writeReplace} method returns
 * an instance of {@code SerializedLambda}, rather than allowing default
 * serialization to proceed.
 *
 * <p>{@code SerializedLambda} has a {@code readResolve} method that looks for
 * a (possibly private) static method called
 * {@code $deserializeLambda$(SerializedLambda)} in the capturing class, invokes
 * that with itself as the first argument, and returns the result.  Lambda classes
 * implementing {@code $deserializeLambda$} are responsible for validating
 * that the properties of the {@code SerializedLambda} are consistent with a
 * lambda actually captured by that class.
 *
 * <p>The identity of a function object produced by deserializing the serialized
 * form is unpredictable, and therefore identity-sensitive operations (such as
 * reference equality, object locking, and {@code System.identityHashCode()} may
 * produce different results in different implementations, or even upon
 * different deserializations in the same implementation.
 *
 * @see LambdaMetafactory
 * @since 1.8
 */

通过查看字节码发现原方法中增加$deserializeLambda$方法,字节码如下,注意是字节码第50行出现了调用lambda的invokedynamic字节码,而50行之前的字节码通过不断调用invokevirtual获取SerializedLambda的各种属性,并使用equals方法对获取到的属性做校验。根据SerializedLambda类注释的关键字在结合字节码可知$deserializeLambda$主要做校验使用。

  0 aload_0
  1 invokevirtual #28 <java/lang/invoke/SerializedLambda.getImplMethodName : ()Ljava/lang/String;>
  4 astore_1
  5 iconst_m1
  6 istore_2
  7 aload_1
  8 invokevirtual #29 <java/lang/String.hashCode : ()I>
 11 lookupswitch 1
    -75308287:  28 (+17)
    default:  39 (+28)
 28 aload_1
 29 ldc #14 <getName>
 31 invokevirtual #30 <java/lang/String.equals : (Ljava/lang/Object;)Z>
 34 ifeq 39 (+5)
 37 iconst_0
 38 istore_2
 39 iload_2
 40 lookupswitch 1
    0:  60 (+20)
    default:  134 (+94)
 60 aload_0
 61 invokevirtual #31 <java/lang/invoke/SerializedLambda.getImplMethodKind : ()I>
 64 iconst_5
 65 if_icmpne 134 (+69)
 68 aload_0
 69 invokevirtual #32 <java/lang/invoke/SerializedLambda.getFunctionalInterfaceClass : ()Ljava/lang/String;>
 72 ldc #33 <com/yousheng/lambda/test/Func>
 74 invokevirtual #34 <java/lang/Object.equals : (Ljava/lang/Object;)Z>
 77 ifeq 134 (+57)
 80 aload_0
 81 invokevirtual #35 <java/lang/invoke/SerializedLambda.getFunctionalInterfaceMethodName : ()Ljava/lang/String;>
 84 ldc #11 <apply>
 86 invokevirtual #34 <java/lang/Object.equals : (Ljava/lang/Object;)Z>
 89 ifeq 134 (+45)
 92 aload_0
 93 invokevirtual #36 <java/lang/invoke/SerializedLambda.getFunctionalInterfaceMethodSignature : ()Ljava/lang/String;>
 96 ldc #12 <(Ljava/lang/Object;)Ljava/lang/Object;>
 98 invokevirtual #34 <java/lang/Object.equals : (Ljava/lang/Object;)Z>
101 ifeq 134 (+33)
104 aload_0
105 invokevirtual #37 <java/lang/invoke/SerializedLambda.getImplClass : ()Ljava/lang/String;>
108 ldc #13 <com/yousheng/lambda/entity/Child>
110 invokevirtual #34 <java/lang/Object.equals : (Ljava/lang/Object;)Z>
113 ifeq 134 (+21)
116 aload_0
117 invokevirtual #38 <java/lang/invoke/SerializedLambda.getImplMethodSignature : ()Ljava/lang/String;>
120 ldc #15 <()Ljava/lang/String;>
122 invokevirtual #34 <java/lang/Object.equals : (Ljava/lang/Object;)Z>
125 ifeq 134 (+9)
128 invokedynamic #2 <apply, BootstrapMethods #0>   // 注意
133 areturn
134 new #39 <java/lang/IllegalArgumentException>
137 dup
138 ldc #40 <Invalid lambda deserialization>
140 invokespecial #41 <java/lang/IllegalArgumentException.<init> : (Ljava/lang/String;)V>
143 athrow

再看一下SerializedLambda中的readResolve方法,通过capturingClass获取$deserializeLambda$方法,最后在进行调用。

private Object readResolve() throws ReflectiveOperationException {
    try {
    Method deserialize = AccessController.doPrivileged(new PrivilegedExceptionAction<>() {
        @Override
        public Method run() throws Exception {
            Method m = capturingClass.getDeclaredMethod("$deserializeLambda$", SerializedLambda.class);
            m.setAccessible(true);
            return m;
        }
    });

    return deserialize.invoke(null, this);
}
catch (PrivilegedActionException e) {
    Exception cause = e.getException();
    if (cause instanceof ReflectiveOperationException)
        throw (ReflectiveOperationException) cause;
    else if (cause instanceof RuntimeException)
        throw (RuntimeException) cause;
    else
        throw new RuntimeException("Exception in SerializedLambda.readResolve", e);
}
}

最后在整理下整个lambda序列化流程,首先是对应的lambda表达式必须实现Serializable接口,在实现Serializable接口后,jvm运行时候会在lambda生成的静态内部类中增加writeReplace方法并在调用的类中增加$deserializeLambda$方法校验使用,在序列化过程中ObjectOutputStream会调用writeReplace方法,将整个lambda表达式转换成SerializedLambda,最后将SerializedLambda类序列化保存。
最后明白了lambda的序列化过程后可以用一个例子模拟lambda的反序列化的过程,首先序列化的对象并非是原来的lambda表达式,而是SerializedLambda对象,通过调用$deserializeLambda$方法生成校验SerializedLambda方队,校验通过则调用对应的Booststrap方法进行一系列转化继续来到LambdaMetaFactory的altMetafactory方法并最终在此方法生成最后的调用CallSite

class LambdaSerialized {
    void serializable() {
        Function<Child, String> function = (Function<Child, String> & Serializable) (Child child) -> {
            System.out.println("test");
            return child.getName();
        };
        System.out.println(Arrays.toString(function.getClass().getDeclaredMethods()));
    }
    
    public static void main(String[] args) throws ClassNotFoundException, InvocationTargetException, NoSuchMethodException, IllegalAccessException {
        Child child = new Child("yousheng", 18);
         // 下方输出结果 yousheng 
        System.out.println(new LambdaSerialized().<Function<Child, String>>test().apply(child));
       
    }
    static <T> T test() throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException, IllegalAccessException {
        SerializedLambda serializedLambda = new SerializedLambda(LambdaSerialized.class, "java/util/function/Function", "apply", "(Ljava/lang/Object;)Ljava/lang/Object;", 5, "com/yousheng/lambda/entity/Child", "getName", "()Ljava/lang/String;", "(Lcom/yousheng/lambda/entity/Child;)Ljava/lang/String;", new Object[0]);
        Method m = Class.forName(serializedLambda.getCapturingClass().replace('/', '.')).getDeclaredMethod("$deserializeLambda$", SerializedLambda.class);
        m.setAccessible(true);
        return (T)m.invoke(null, serializedLambda);
    }
}
// Child类定义如下
class Child {
    private String name;
    private int age;
}

image.png
查看栈帧方法由$deserializeLambda$调用至 ,
image.png
最后调用到事先定义好的方法。
image.png

相关文章
|
14天前
|
Java
探索Java中的Lambda表达式
【10月更文挑战第37天】本文将带你深入理解Java的Lambda表达式,从基础语法到高级特性,通过实例讲解其在函数式编程中的应用。我们还将探讨Lambda表达式如何简化代码、提高开发效率,并讨论其在实际项目中的应用。
|
16天前
|
Java API
Java中的Lambda表达式与函数式编程####
【10月更文挑战第29天】 本文将深入探讨Java中Lambda表达式的实现及其在函数式编程中的应用。通过对比传统方法,我们将揭示Lambda如何简化代码、提高可读性和维护性。文章还将展示一些实际案例,帮助读者更好地理解和应用Lambda表达式。 ####
|
16天前
|
JSON 自然语言处理 Java
这款轻量级 Java 表达式引擎,真不错!
AviatorScript 是一个高性能、轻量级的脚本语言,基于 JVM(包括 Android 平台)。它支持数字、字符串、正则表达式、布尔值等基本类型,以及所有 Java 运算符。主要特性包括函数式编程、大整数和高精度运算、完整的脚本语法、丰富的内置函数和自定义函数支持。适用于规则判断、公式计算、动态脚本控制等场景。
|
21天前
|
Java API 开发者
Java中的Lambda表达式与函数式编程####
在Java的演变过程中,Lambda表达式和函数式编程的引入无疑是一次重大的飞跃。本文将深入探讨Lambda表达式的定义、用法及优势,并结合实例说明如何在Java中利用Lambda表达式进行函数式编程。通过对比传统编程方式,揭示Lambda表达式如何简化代码、提高开发效率和可维护性。 ####
|
29天前
|
自然语言处理 安全 Java
Aviator Java 表达式引擎
AviatorScript 是一门高性能、轻量级寄宿于 JVM 之上的脚本语言。
48 10
|
27天前
|
Java API 数据处理
探索Java中的Lambda表达式与Stream API
【10月更文挑战第22天】 在Java编程中,Lambda表达式和Stream API是两个强大的功能,它们极大地简化了代码的编写和提高了开发效率。本文将深入探讨这两个概念的基本用法、优势以及在实际项目中的应用案例,帮助读者更好地理解和运用这些现代Java特性。
|
1月前
|
Java API
[Java]Lambda表达式
本文主要介绍了Java中的Lambda表达式,包括其优化匿名内部类的方式、使用规范、内置函数式接口及方法引用等内容。文章详细解析了Lambda的基础语法、参数列表、方法体的简化规则,以及如何利用Lambda优化代码。此外,还探讨了Lambda的作用域和引用规则,强调了对局部变量、成员变量和常量的访问限制,旨在帮助读者全面理解和掌握Lambda表达式的应用。
19 0
[Java]Lambda表达式
|
1月前
|
Java 程序员 API
Java中的Lambda表达式:简化代码的秘密武器
【10月更文挑战第11天】 在Java编程中,Lambda表达式是一种简洁而强大的工具,它允许我们将函数作为参数传递给其他方法。本文将介绍Lambda表达式的基本概念、使用方法以及在实际项目中的应用案例,帮助你更好地理解和利用这一特性来简化代码。
24 8
|
1月前
|
Java 测试技术 开发者
🌟Java零基础-Lambda运算符详解 🌟
【10月更文挑战第12天】本文收录于「滚雪球学Java」专栏,专业攻坚指数级提升,希望能够助你一臂之力,帮你早日登顶实现财富自由🚀;同时,欢迎大家关注&&收藏&&订阅!持续更新中,up!up!up!!
23 1
下一篇
无影云桌面