react动态form表单 Reactor Flux动态数据注入与多源合并策略

论文在Reactor框架中进行了讨论，如何向一个由外部库提供的现有Flux动态注入数据，以及如何将自定义数据流与外部Flux进行有效的合并。文章将详细介绍如何利用Sinks创建可控的发射器，并通过Flux.merge()等操作符将多个数据源整合，同时会针对UnicastProcessor等瞬时订阅源的特殊情况提供解决方案和注意事项。

在响应式编程中，Flux代表一个0到N个元素的异步序列。当由外部库提供、我们无法直接控制其数据面向发射机制的 Flux 时，如何将我们自己的数据注入其中，或者将其与我们自己的数据流结合，是一个常见的挑战。通常，Flux 本身不提供直接的发射方法，因为它的设计理念是作为数据据流的消费者不是直接的生产者（对于外部数据源而言）。挑战：向现有Flux注入数据

假设我们有一个外部库方法，它返回一个Fluxlt；MappedTypegt；：Fluxlt；MappedTypegt；通量图 = Library.createMappingToMappedType()；登录后复制

我们希望能够将自己的原始对象（例如myObj）发送到这个aFluxMap中，让它们被处理并转换为MappedType，然后继续后续操作。观上，我们可能期望有一个类似aFluxMap.emit(myObj)的方法，但这样的方法在Flux或Mono中并不存在。

一种常见的误解是尝试使用FluxProcessor和FluxSink来解决：FluxProcessor p = UnicastProcessor.create().serialize()；FluxSink sink = p.sink()；sink.next(mess)； // 这会将消息发送到新创建的 Flux 'p'登录后复制

这种方法的问题同样，它创建了一个新的 Flux (p) 传输其发送消息，而不是仅仅将消息发送到我们已有的aFluxMap。我们仅仅需要我们自己的数据流与aFluxMap关联起来。解决方案：合并数据流

Reactor框架提供了强大的操作符来组合和合并不同的数据据流。解决上述问题的核心思路是：创建一个我们自己可以控制的Flux，用于发送我们的自定义数据。使用合并操作符（如merge、concat、zip等）将这个自定义Flux与外部库提供的aFluxMap结合起来。1. 创建可控的自定义数据流

在Reactor 3.4及更高版本中，推荐使用Sinks API来创建可控的发送器。Sinks.many()创建可以一个多值的发送器，提供了tryEmitNext、tryEmitComplete等方法来安全地发送数据。

importreactor.core.publisher.Flux；importreactor.core.publisher.Sinks；//假设MappedType是一个已定义的类型class MappedType { private String value； //构造函数、getter等 public MappedType(String value) { this.value = value； } public String getValue() { return value； } @Override public String toString() { return quot；MappedType(quot； value quot；)quot；； }}public class CustomEmitter { private final Sinks.Manylt；Stringgt； myDataSink = Sinks.many().multicast().onBackPressureBuffer()； private final Fluxlt；Stringgt； myDataFlux = myDataSink.asFlux()； // 模拟外部库的Flux public Fluxlt；MappedTypegt； createExternalFlux() { return Flux.just(quot；External1quot；， quot；External2quot；) .map(s -gt；新MappedType(quot；External-quot； s))； } // 发送自定义数据的方法 public void emitMyData(String data) { myDataSink.tryEmitNext(data)； } public Fluxlt；Stringgt； getMyDataFlux() { return myDataFlux； } public static void main(String[] args) { CustomEmitteremitter = new CustomEmitter()； // 1. 获取外部库的Flux Fluxlt；MappedTypegt； aFluxMap = emitter.createExternalFlux()； // 2.创建我们自己的Flux（这里我们直接使用原始字符串，稍后进行转换） Fluxlt；Stringgt； customRawDataFlux = emitter.getMyDataFlux()； // 3.将自定义原始数据转换为MappedType Fluxlt；MappedTypegt； customMappedDataFlux = customRawDataFlux .map(raw -gt； {

System.out.println(quot；转换自定义原始数据：quot；raw)； // 模拟转换逻辑 return new MappedType(quot；Custom-quot； raw)； })； // 4. 合并两个Flux Fluxlt；MappedTypegt；combinedFlux = Flux.merge(aFluxMap，customMappedDataFlux)； // 订阅合并后的Flux并处理数据combinedFlux.doOnNext(converted -gt； System.out.println(quot；已接收： quot；converted)) .subscribe( null， // onNext 错误 -gt； System.err.println(quot；Error： quot； error)， () -gt； System.out.println(quot；组合通量完成.quot；) )； // 动态发送自定义数据emitter.emitMyData(quot；MyDataAquot；)；emitter.emitMyData(quot；MyDataBquot；)； //模拟完成外部Flux完成后，手动自定义Flux // 如果不手动完成，程序会一直运行等待更多数据 // myDataSink.tryEmitComplete()； // 实际应用中业务根据逻辑决定何时完成}}登录后复制允许

在上述例子中：我们创建了一个Sinks.Manylt；Stringgt；来作为自定义数据源。emitMyData方法我们随时向这个Sink发送数据。customMappedDataFlux负责将我们发送的原始String数据转换为MappedType。Flux.merge(aFluxMap， customMappedDataFlux)将外部Flux和我们自己的Flux合并成一个单一的Flux。合并操作符会从两个源接收元素，并按照它们到的顺序发送。2. 常用合并操作符Flux.merge(Publisher...sources)：并发地合并多个Publisher的元素，元素到达的顺序是它们在输出Flux中的顺序。适用于对顺序不敏感但决定快速处理所有数据的场景。Flux.concat(Publisher...sources)：需要顺序地连接多个Publisher。它会等待一个Publisher完成之前，然后才订阅下一个Publisher。适用于需要严格保持数据源顺序的场景。

Flux.zip(Publisher...sources，Functionlt；Object[]，Ogt；combinator)：将多个Publisher的元素两两配对，并使用提供的组合器函数将它们组合成一个新的元素。它会等待所有源都发出一个元素后才进行组合。适用于需要将不同类型或不同来源的数据关联起来的场景。

根据具体需求选择合适的合并操作符。在大多数动态填充数据的场景中，合并是最常用的，因为它允许并发处理来自不同源的数据。注意事项与常见问题1. UnicastProcessor的订阅限制

在原始问题中提到了一个重要的更新：Library.createMappingToMappedType()返回的aFluxMap的内部源可能是一个UnicastProcessor，并且该UnicastProcessor可能已经被库内部订阅。当尝试通过p.flatMap(raw -gt；aFluxMap).subscribe()再次订阅aFluxMap时，会遇到“UnicastProcessor can be subscribe”

解释：UnicastProcessor是一个单播处理器，它只允许一个订阅者。一旦被订阅，它就会开始发送数据。如果尝试第二次订阅，就会抛出异常。

解决方案：理解合并的工作方式： Flux.merge(aFluxMap， customMappedDataFlux)操作通常符对aFluxMap进行一次订阅。这意味着如果aFluxMap本身是一个有效的Flux，即使其内部使用了UnicastProcessor，但它外部暴露的接口只能成功订阅一次），那么merge操作应该能够。避免重复订阅：原始问题中的p.flatMap(raw -gt； aFluxMap)会导致对aFluxMap进行多次订阅（每次发送一个raw，就会尝试订阅一次aFluxMap），这就是导致UnicastProcessor报错的原因。务必避免在flatMap等操作中对单播源进行重复订阅。如果aFluxMap本身就是已订阅的UnicastProcessor：如果Library.createMappingToMappedType()返回的Flux本身就是一个已经启动并被订阅过的UnicastProcessor，那么任何其一的再次订阅都会失败。在这种极端情况下，你不可能直接“合并”它，而只能将它作为一个消费者（即，只订阅一次）。应对策略：如果aFluxMap是一个已订阅且不能再次订阅的单播源，你无法通过merge来将你的数据“注入”到它的上游。你只能将你的数据与aFluxMap的输出进行合并。这意味着aFluxMap的数据流是独立的，你的数据流也是独立的，它们在下游汇合。上述Flux.merge(aFluxMap，customMappedDataFlux)的方案是这样的，将会是两个独立的、可订阅的Flux合并。

2. 背压（BackPressure）

在使用Sinks.many().multicast().onBackPressureBuffer()时，我们选择了onBackPressureBuffer策略，这意味着如果下游消费者处理速度慢于上游传输速度，Sink会缓冲元素。根据实际需求，你可能需要选择其他背压策略，例如onBackPressureDrop（丢弃元素）或onBackPressureError（发布错误）。3. 资源管理与完成信号

当你的自定义数据流不再有数据需要发送时，记得调用myDataSink.tryEmitComplete()来发出完成信号。这对于下游的Flux操作符（比如合并来说）来说很重要，它们需要知道所有上游源都什么时候完成，以便自己也能完成。总结

在Reactor中，向一个由外部库提供的现有Flux动态填充数据据此，并不是通过直接的emit方法实现，而是通过创建一个可控的自定义Flux，然后利用Flux.merge()等操作符将其与外部Flux合并。这种模式遵循了响应式编程的原则，即通过组合和转换数据流来构建复杂的业务逻辑。同时，需要特别注意UnicastProcessor等单订阅源的特性，避免正确的重复订阅操作。

以上就是Reactor Flux动态数据注入与多源合并策略的详细内容，更多请关注乐哥常识网其他相关文章！