Open ypk1226 opened 4 years ago
Flux<String> flux = Flux.generate(
() -> 0,
(state, sink) -> {
sink.next("3 x " + state + " = " + 3 * state);
if (state == 5) sink.complete();
return state + 1;
});
没看懂,不是不懂逻辑。是不懂sink对象是默认的吗?()->[] 换成数组会怎么样?规范中的Processor 接口用哪里了?
@swit1983
Flux<String> flux = Flux.generate( () -> 0, (state, sink) -> { sink.next("3 x " + state + " = " + 3 * state); if (state == 5) sink.complete(); return state + 1; });
没看懂,不是不懂逻辑。是不懂sink对象是默认的吗?()->[] 换成数组会怎么样?规范中的Processor 接口用哪里了?
没想到会有评论,所以很久没看,回复的晚了,哈哈。
public static <T, S> Flux<T> generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator) {
return onAssembly((Flux)(new FluxGenerate(stateSupplier, generator)));
}
看这个方法的签名,lambda把类型隐藏了,SynchronousSink这个类型就是处理消息节点的。
规范中的Processor在实际使用中根本看不到了,被封装的严严实实,那个方法的真实处理,被实现框架封装好了,它很复杂。
http://www.ypk1226.com/2019/08/02/reactive/project-reactor/
上一篇文章中,我们介绍了Reactive Streams规范,现在学习一个Reactive Streams规范的流行实现:Project Reactor的核心项目Reactor Core。