From 40726b853e60296e2e82cf8d0a16d5e8db5cc878 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Youth=EF=BC=8E=E9=9C=96?= Date: Wed, 17 Jul 2019 20:48:12 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=AD=A3=E4=B8=A4=E4=B8=AA=E8=AF=8D?= =?UTF-8?q?=E8=AF=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 乘放 -> 盛放 子类重载 Sink -> 子类重写 Sink --- 6-Stream Pipelines.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/6-Stream Pipelines.md b/6-Stream Pipelines.md index 5fcaf3b..c3b091d 100644 --- a/6-Stream Pipelines.md +++ b/6-Stream Pipelines.md @@ -89,7 +89,7 @@ Stream流水线组织结构示意图如下：
方法名作用
void begin(long size)开始遍历元素之前调用该方法，通知Sink做好准备。
void end()所有元素遍历完成之后调用，通知Sink没有更多的元素了。
boolean cancellationRequested()是否可以结束操作，可以让短路操作尽早结束。
void accept(T t)遍历元素时调用，接受一个待处理元素，并对元素进行处理。Stage把自己包含的操作和回调方法封装到该方法里，前一个Stage只需要调用当前Stage.accept(T t)方法就行了。
-有了上面的协议，相邻Stage之间调用就很方便了，每个Stage都会将自己的操作封装到一个Sink里，前一个Stage只需调用后一个Stage的`accept()`方法即可，并不需要知道其内部是如何处理的。当然对于有状态的操作，Sink的`begin()`和`end()`方法也是必须实现的。比如Stream.sorted()是一个有状态的中间操作，其对应的Sink.begin()方法可能创建一个乘放结果的容器，而accept()方法负责将元素添加到该容器，最后end()负责对容器进行排序。对于短路操作，`Sink.cancellationRequested()`也是必须实现的，比如Stream.findFirst()是短路操作，只要找到一个元素，cancellationRequested()就应该返回*true*，以便调用者尽快结束查找。Sink的四个接口方法常常相互协作，共同完成计算任务。**实际上Stream API内部实现的的本质，就是如何重载Sink的这四个接口方法**。 +有了上面的协议，相邻Stage之间调用就很方便了，每个Stage都会将自己的操作封装到一个Sink里，前一个Stage只需调用后一个Stage的`accept()`方法即可，并不需要知道其内部是如何处理的。当然对于有状态的操作，Sink的`begin()`和`end()`方法也是必须实现的。比如Stream.sorted()是一个有状态的中间操作，其对应的Sink.begin()方法可能创建一个盛放结果的容器，而accept()方法负责将元素添加到该容器，最后end()负责对容器进行排序。对于短路操作，`Sink.cancellationRequested()`也是必须实现的，比如Stream.findFirst()是短路操作，只要找到一个元素，cancellationRequested()就应该返回*true*，以便调用者尽快结束查找。Sink的四个接口方法常常相互协作，共同完成计算任务。**实际上Stream API内部实现的的本质，就是如何重写Sink的这四个接口方法**。 有了Sink对操作的包装，Stage之间的调用问题就解决了，执行时只需要从流水线的head开始对数据源依次调用每个Stage对应的Sink.{begin(), accept(), cancellationRequested(), end()}方法就可以了。一种可能的Sink.accept()方法流程是这样的：