以编程方式创建序列
目录
本节介绍通过以编程方式定义 Flux 或 Mono 相关的事件（onNext、onError 和 onComplete）来创建它们。所有这些方法的共同点是，它们都暴露一个 API 来触发事件，我们称之为 sink。实际上有几种 sink 变体，我们稍后就会介绍。
1. 同步 generate
以编程方式创建 Flux 最简单的形式是通过 generate 方法，它接受一个生成器函数。
这适用于同步且逐个的发射，也就是说 sink 是一个 SynchronousSink，其 next() 方法在每次回调调用中至多只能调用一次。之后您还可以额外调用 error(Throwable) 或 complete()，但这是可选的。
最有用的变体大概是那个还让您维护一个状态的版本，您可以在使用 sink 时引用该状态来决定下一步发射什么。此时生成器函数变为 BiFunction<S, SynchronousSink<T>, S>，其中 <S> 是状态对象的类型。您必须为初始状态提供一个 Supplier<S>，而您的生成器函数现在每一轮都返回一个新状态。
例如，您可以用一个 int 作为状态：
Flux<String> flux = Flux.generate(
() -> 0, 1
(state, sink) -> {
sink.next("3 x " + state + " = " + 3*state); 2
if (state == 10) sink.complete(); 3
return state + 1; 4
});
我们提供初始状态值 0。
我们用状态来选择发射什么（3 的乘法表中的一行）。
我们还用它来选择何时停止。
我们返回一个新状态，供下一次调用使用（除非序列在本次调用中已终止）。
前述代码生成 3 的乘法表，得到如下序列：
3 x 0 = 0
3 x 1 = 3
3 x 2 = 6
3 x 3 = 9
3 x 4 = 12
3 x 5 = 15
3 x 6 = 18
3 x 7 = 21
3 x 8 = 24
3 x 9 = 27
3 x 10 = 30
您也可以使用一个可变的 <S>。例如，上面的例子可以改写成用单个 AtomicLong 作为状态，并在每一轮修改它：
Flux<String> flux = Flux.generate(
AtomicLong::new, 1
(state, sink) -> {
long i = state.getAndIncrement(); 2
sink.next("3 x " + i + " = " + 3*i);
if (i == 10) sink.complete();
return state; 3
});
这一次，我们生成一个可变对象作为状态。
我们在这里修改状态。
我们返回同一个实例作为新状态。
如果您的状态对象需要清理一些资源，请使用 generate(Supplier<S>, BiFunction, Consumer<S>) 变体来清理最后一个状态实例。
以下示例使用包含 Consumer 的 generate 方法：
Flux<String> flux = Flux.generate(
AtomicLong::new,
(state, sink) -> { 1
long i = state.getAndIncrement(); 2
sink.next("3 x " + i + " = " + 3*i);
if (i == 10) sink.complete();
return state; 3
}, (state) -> System.out.println("state: " + state)); 4
同样，我们生成一个可变对象作为状态。
我们在这里修改状态。
我们返回同一个实例作为新状态。
我们在这个
Consumerlambda 的输出中看到最后一个状态值（11）。
当状态包含一个数据库连接或其他需要在过程结束时处理的资源时，该 Consumer lambda 可以关闭连接，或以其他方式处理任何应在过程结束时完成的任务。
2. 异步与多线程：create
create 是一种更高级的以编程方式创建 Flux 的形式，它适合每一轮多次发射，甚至可以来自多个线程。
它暴露一个 FluxSink，带有其 next、error 和 complete 方法。与 generate 不同，它没有基于状态的变体。另一方面，它可以在回调中触发多线程事件。
create 在把已有 API 与响应式世界桥接时非常有用——例如基于监听器的异步 API。
create 不会并行化您的代码，也不会使代码变为异步，尽管它可以与异步 API 配合使用。如果您在 create lambda 内部阻塞，就会让自己面临死锁及类似副作用的风险。即使使用 subscribeOn，也要注意：一个长时间阻塞的 create lambda（例如调用 sink.next(t) 的无限循环）可能会锁死整条管线：由于循环耗尽了本应运行这些请求的同一线程，请求将永远无法被执行。请使用 subscribeOn(Scheduler, false) 变体：requestOnSeparateThread = false 会用 Scheduler 线程来运行 create，同时仍通过在原始线程中执行 request 来让数据流动。
设想您使用一个基于监听器的 API。它按数据块处理数据，并有两个事件：(1) 一个数据块已就绪；(2) 处理完成（终止事件），如 MyEventListener 接口所表示的那样：
interface MyEventListener<T> {
void onDataChunk(List<T> chunk);
void processComplete();
}
您可以使用 create 将其桥接为一个 Flux<T>：
Flux<String> bridge = Flux.create(sink -> {
myEventProcessor.register( 4
new MyEventListener<String>() { 1
public void onDataChunk(List<String> chunk) {
for(String s : chunk) {
sink.next(s); 2
}
}
public void processComplete() {
sink.complete(); 3
}
});
});
桥接到
MyEventListenerAPI一个数据块中的每个元素都成为
Flux中的一个元素。processComplete事件被转换为onComplete。所有这些都在
myEventProcessor执行时异步完成。
此外，由于 create 可以桥接异步 API 并管理背压，您还可以通过指定一个 OverflowStrategy 来细化背压方面的行为：
-
IGNORE，完全忽略下游的背压请求。当下游队列满时，这可能产生IllegalStateException。 -
ERROR，当下游跟不上时发出一个IllegalStateException信号。 -
DROP，当下游尚未准备好接收时丢弃传入信号。 -
LATEST，让下游只获取上游最新的信号。 -
BUFFER（默认），当下游跟不上时缓冲所有信号。（这会进行无界缓冲，并可能导致OutOfMemoryError）。
Mono 也有一个 create 生成器。Mono 的 create 所用的 MonoSink 不允许多次发射。它会丢弃第一个信号之后的所有信号。
3. 异步但单线程：push
push 是 generate 与 create 之间的折中，适合处理来自单个生产者的事件。它与 create 相似之处在于：它也可以是异步的，并且可以使用 create 支持的任何一种溢出策略来管理背压。然而，一次只能有一个生产线程调用 next、complete 或 error。
Flux<String> bridge = Flux.push(sink -> {
myEventProcessor.register(
new SingleThreadEventListener<String>() { 1
public void onDataChunk(List<String> chunk) {
for(String s : chunk) {
sink.next(s); 2
}
}
public void processComplete() {
sink.complete(); 3
}
public void processError(Throwable e) {
sink.error(e); 4
}
});
});
桥接到
SingleThreadEventListenerAPI。事件通过单个监听器线程用
next推入 sink。complete事件也由同一个监听器线程产生。error事件同样由同一个监听器线程产生。
3.1. 混合 push/pull 模型
大多数 Reactor 操作符（如 create）遵循一种混合的**推/拉（push/pull）**模型。我们的意思是：尽管大部分处理是异步的（暗示一种推的方式），其中仍有一个小小的拉的成分：请求（request）。
消费者从源拉取数据，意思是它在首次被请求之前不会发射任何东西。源则在数据可用时随时向消费者推送数据，但不会超出其请求的数量。
请注意，push() 和 create() 都允许设置一个 onRequest 处理器，以便管理请求量，并确保只有在有待处理请求时才通过 sink 推送数据。
Flux<String> bridge = Flux.create(sink -> {
myMessageProcessor.register(
new MyMessageListener<String>() {
public void onMessage(List<String> messages) {
for(String s : messages) {
sink.next(s); 3
}
}
});
sink.onRequest(n -> {
List<String> messages = myMessageProcessor.getHistory(n); 1
for(String s : messages) {
sink.next(s); 2
}
});
});
在收到请求时轮询消息。
如果消息立即可用，就把它们推送到 sink。
之后异步到达的其余消息也会被投递。
3.2. 在 push() 或 create() 之后清理
两个回调 onDispose 和 onCancel 用于在取消或终止时执行清理。onDispose 可用于在 Flux 完成、出错或被取消时执行清理。onCancel 可用于在 onDispose 清理之前执行任何专门针对取消的操作。
Flux<String> bridge = Flux.create(sink -> {
sink.onRequest(n -> channel.poll(n))
.onCancel(() -> channel.cancel()) 1
.onDispose(() -> channel.close()) 2
});
onCancel最先被调用，仅针对取消信号。onDispose针对完成、错误或取消信号被调用。
4. handle
handle 方法略有不同：它是一个实例方法，意味着它被链接到一个已有源之上（与常见操作符一样）。它在 Mono 和 Flux 中都存在。
它与 generate 相近，意思是它使用一个 SynchronousSink，并且只允许逐个发射。然而，handle 可以用来从每个源元素派生出一个任意值，甚至可能跳过某些元素。这样一来，它可以充当 map 与 filter 的组合。handle 的签名如下：
Flux<R> handle(BiConsumer<T, SynchronousSink<R>>);
让我们看一个例子。响应式流规范不允许序列中出现 null 值。如果您想执行一个 map，但又想把一个已有的方法用作 map 函数，而该方法有时会返回 null，该怎么办？
例如，下面这个方法可以安全地应用于一个整数源：
public String alphabet(int letterNumber) {
if (letterNumber < 1 || letterNumber > 26) {
return null;
}
int letterIndexAscii = 'A' + letterNumber - 1;
return "" + (char) letterIndexAscii;
}
然后我们可以用 handle 来移除任何 null：
Flux<String> alphabet = Flux.just(-1, 30, 13, 9, 20)
.handle((i, sink) -> {
String letter = alphabet(i); 1
if (letter != null) 2
sink.next(letter); 3
});
alphabet.subscribe(System.out::println);
映射为字母。
如果这个“map 函数”返回 null……
通过不调用
sink.next将其过滤掉。
这会打印出：
M
I
T