Spring Reactor 项目核心库Reactor Core
Reactor Core
Non-Blocking Reactive Streams Foundation for the JVM both implementing a Reactive Extensions inspired API and efficient event streaming support.
Getting it
Reactor 3 requires Java 8 or + to run.
With Gradle from repo.spring.io or Maven Central repositories (stable releases only):
repositories {
// maven { url 'http://repo.spring.io/snapshot' }
maven { url 'http://repo.spring.io/milestone' }
mavenCentral()
}
dependencies {
//compile "io.projectreactor:reactor-core:3.1.4.RELEASE"
//testCompile("io.projectreactor:reactor-test:3.1.4.RELEASE")
compile "io.projectreactor:reactor-core:3.2.0.M1"
testCompile("io.projectreactor:reactor-test:3.2.0.M1")
}
See the reference documentation for more information on getting it (eg. using Maven, or on how to get milestones and snapshots).
Note about Android support: Reactor 3 doesn't officially support nor target Android. However it should work fine with Android SDK 26 (Android O) and above. See the complete note in the reference guide.
Getting Started
New to Reactive Programming or bored of reading already ? Try the Introduction to Reactor Core hands-on !
If you are familiar with RxJava or if you want to check more detailled introduction, be sure to check https://www.infoq.com/articles/reactor-by-example !
Flux
A Reactive Streams Publisher with basic flow operators.
- Static factories on Flux allow for source generation from arbitrary callbacks types.
- Instance methods allows operational building, materialized on each Flux#subscribe(), Flux#subscribe() or multicasting operations such as Flux#publish and Flux#publishNext.
Flux in action :
Flux.fromIterable(getSomeLongList())
.mergeWith(Flux.interval(100))
.doOnNext(serviceA::someObserver)
.map(d -> d * 2)
.take(3)
.onErrorResumeWith(errorHandler::fallback)
.doAfterTerminate(serviceM::incrementTerminate)
.subscribe(System.out::println);
Mono
A Reactive Streams Publisher constrained to ZERO or ONE element with appropriate operators.
- Static factories on Mono allow for deterministic zero or one sequence generation from arbitrary callbacks types.
- Instance methods allows operational building, materialized on each Mono#subscribe() or Mono#get() eventually called.
Mono in action :
Mono.fromCallable(System::currentTimeMillis)
.flatMap(time -> Mono.first(serviceA.findRecent(time), serviceB.findRecent(time)))
.timeout(Duration.ofSeconds(3), errorHandler::fallback)
.doOnSuccess(r -> serviceM.incrementSuccess())
.subscribe(System.out::println);
Blocking Mono result :
Tuple2<Long, Long> nowAndLater =
Mono.zip(
Mono.just(System.currentTimeMillis()),
Flux.just(1).delay(1).map(i -> System.currentTimeMillis()))
.block();
Schedulers
Reactor uses a Scheduler as a contract for arbitrary task execution. It provides some guarantees required by Reactive Streams flows like FIFO execution.
You can use or create efficient schedulers to jump thread on the producing flows (subscribeOn) or receiving flows (publishOn):
Mono.fromCallable( () -> System.currentTimeMillis() )
.repeat()
.publishOn(Schedulers.single())
.log("foo.bar")
.flatMap(time ->
Mono.fromCallable(() -> { Thread.sleep(1000); return time; })
.subscribeOn(Schedulers.parallel())
, 8) //maxConcurrency 8
.subscribe();
ParallelFlux
ParallelFlux can starve your CPU's from any sequence whose work can be subdivided in concurrent
tasks. Turn back into a Flux
with ParallelFlux#sequential()
, an unordered join or
use abitrary merge strategies via 'groups()'.
Mono.fromCallable( () -> System.currentTimeMillis() )
.repeat()
.parallel(8) //parallelism
.runOn(Schedulers.parallel())
.doOnNext( d -> System.out.println("I'm on thread "+Thread.currentThread()) )
.subscribe()
Custom sources : Flux.create and FluxSink, Mono.create and MonoSink
To bridge a Subscriber or Processor into an outside context that is taking care of
producing non concurrently, use Flux#create
, Mono#create
.
Flux.create(sink -> {
ActionListener al = e -> {
sink.next(textField.getText());
};
// without cancellation support:
button.addActionListener(al);
// with cancellation support:
sink.onCancel(() -> {
button.removeListener(al);
});
},
// Overflow (backpressure) handling, default is BUFFER
FluxSink.OverflowStrategy.LATEST)
.timeout(3)
.doOnComplete(() -> System.out.println("completed!"))
.subscribe(System.out::println)
The Backpressure Thing
Most of this cool stuff uses bounded ring buffer implementation under the hood to mitigate signal processing difference between producers and consumers. Now, the operators and processors or any standard reactive stream component working on the sequence will be instructed to flow in when these buffers have free room AND only then. This means that we make sure we both have a deterministic capacity model (bounded buffer) and we never block (request more data on write capacity). Yup, it's not rocket science after all, the boring part is already being worked by us in collaboration with Reactive Streams Commons on going research effort.
What's more in it ?
"Operator Fusion" (flow optimizers), health state observers, helpers to build custom reactive components, bounded queue generator, hash-wheel timer, converters from/to Java 9 Flow, Publisher and Java 8 CompletableFuture. The repository contains a reactor-test
project with test features like the StepVerifier
.
Reference Guide
http://projectreactor.io/docs/core/release/reference/docs/index.html
Javadoc
https://projectreactor.io/docs/core/release/api/
Getting started with Flux and Mono
https://github.com/reactor/lite-rx-api-hands-on
Reactor By Example
https://www.infoq.com/articles/reactor-by-example
Head-First Spring & Reactor
https://github.com/reactor/head-first-reactive-with-spring-and-reactor/
Beyond Reactor Core
- Everything to jump outside the JVM with the non-blocking drivers from Reactor Netty.
- Reactor Addons provide for adapters and extra operators for Reactor 3.
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法