1. 理解 Reactor Flux 的发布者特性
在 Reactor 编程模型中，Flux 和 Mono 是数据发布者（Publisher），它们负责按照 Reactive Streams 规范将数据序列发布给订阅者（Subscriber）。与传统的命令式编程中的队列或列表不同，Flux 并非一个可以直接“写入”或“发射”数据进去的容器。因此，像 aFluxMap.emit(myObj) 这样的方法在 Flux 或 Mono 接口中是不存在的。
如果你希望将自定义数据引入到响应式流中，你需要做的是创建一个 新的 发布者，由这个发布者来产生你的数据。
2. 创建自定义数据源
为了动态地向响应式流中注入数据，Reactor 提供了多种机制来创建可控制的发布者。其中最常用且灵活的方式是使用 Sinks API 或 FluxProcessor。
2.1 使用 Sinks.many() (推荐)
Sinks 是 Reactor 3.4 引入的更现代、更安全的 API，用于创建多值（Sinks.many()）或单值（Sinks.one()）的发布者，并提供了一个 FluxSink 类似的接口来发射数据。
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
public class CustomDataSource {
// 定义一个 Sinks.Many 对象，用于发射 MyRawType 类型的数据
// 这里使用 unicast() 模式，表示只有一个订阅者
private final Sinks.Many<MyRawType> rawTypeSink = Sinks.many().unicast().onBackpressureBuffer();
// 暴露一个 Flux 供外部订阅
public Flux<MyRawType> getRawTypeFlux() {
return rawTypeSink.asFlux();
}
// 外部调用此方法来发射数据
public void emitRawType(MyRawType data) {
rawTypeSink.tryEmitNext(data).orThrow(); // 尝试发射数据，如果失败则抛出异常
}
// 示例：MyRawType 是你的原始数据类型
static class MyRawType {
String id;
// ... constructor, getters, etc.
}
public static void main(String[] args) {
CustomDataSource dataSource = new CustomDataSource();
Flux<MyRawType> myRawFlux = dataSource.getRawTypeFlux();
myRawFlux.map(raw -> {
// 模拟将 MyRawType 转换为 MappedType
System.out.println("Converting raw: " + raw.id);
return new MappedType("Mapped-" + raw.id);
}).subscribe(mapped -> System.out.println("Received MappedType: " + mapped.name));
// 动态发射数据
dataSource.emitRawType(new MyRawType("A"));
dataSource.emitRawType(new MyRawType("B"));
// ...
}
// 示例：MappedType 是外部库期望的类型
static class MappedType {
String name;
// ... constructor, getters, etc.
public MappedType(String name) { this.name = name; }
}
}2.2 使用 FluxProcessor (传统方式)
FluxProcessor 是一类特殊的 Flux，它同时实现了 Subscriber 和 Publisher 接口，可以作为数据处理链中的桥梁。UnicastProcessor 是一个常见的选择，但它有“一次性订阅”的限制（详见后续章节）。
import reactor.core.publisher.Flux;
import reactor.core.publisher.UnicastProcessor;
import reactor.core.publisher.FluxSink;
public class CustomDataSourceProcessor {
private final UnicastProcessor<MyRawType> myProcessor = UnicastProcessor.create();
private final FluxSink<MyRawType> mySink = myProcessor.sink();
public Flux<MyRawType> getRawTypeFlux() {
return myProcessor;
}
public void emitRawType(MyRawType data) {
mySink.next(data);
}
// MyRawType 和 MappedType 定义同上
static class MyRawType { String id; public MyRawType(String id) { this.id = id; } }
static class MappedType { String name; public MappedType(String name) { this.name = name; } }
public static void main(String[] args) {
CustomDataSourceProcessor dataSource = new CustomDataSourceProcessor();
Flux<MyRawType> myRawFlux = dataSource.getRawTypeFlux();
myRawFlux.map(raw -> {
System.out.println("Converting raw: " + raw.id);
return new MappedType("Mapped-" + raw.id);
}).subscribe(mapped -> System.out.println("Received MappedType: " + mapped.name));
dataSource.emitRawType(new MyRawType("X"));
dataSource.emitRawType(new MyRawType("Y"));
}
}3. 合并现有 Flux 与新数据流
一旦你创建了自己的数据源（例如 myRawFlux），下一步就是将其与外部库提供的 Flux<MappedType> 进行整合。这里的关键是，你的自定义数据在与外部库的 Flux<MappedType> 合并之前，通常需要先转换为相同的类型 (MappedType)。
假设外部库提供的方法如下：
public class Library {
public static Flux<MappedType> createMappingToMappedType() {
// 模拟一个持续产生 MappedType 的 Flux
return Flux.just(new MappedType("Lib-1"), new MappedType("Lib-2"))
.delayElements(java.time.Duration.ofMillis(100));
}
}现在，我们将你的自定义数据流（经过转换后）与 Library.createMappingToMappedType() 返回的 Flux 进行合并。
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.time.Duration;
public class FluxMergingExample {
// 假设这是你的原始数据类型和目标映射类型
static class MyRawType { String id; public MyRawType(String id) { this.id = id; } }
static class MappedType { String name; public MappedType(String name) { this.name = name; } }
// 模拟外部库
static class Library {
public static Flux<MappedType> createMappingToMappedType() {
System.out.println("Library.createMappingToMappedType() called.");
return Flux.interval(Duration.ofMillis(200)) // 每200ms产生一个元素
.map(i -> new MappedType("Lib-Item-" + i))
.take(3); // 只取3个元素
}
}
// 模拟将原始类型转换为映射类型的方法
private static MappedType convertRawToMappedType(MyRawType raw) {
System.out.println("Converting raw: " + raw.id);
return new MappedType("My-Converted-" + raw.id);
}
public static void main(String[] args) throws InterruptedException {
// 1. 创建你的自定义数据源
Sinks.Many<MyRawType> myRawSink = Sinks.many().unicast().onBackpressureBuffer();
Flux<MyRawType> myRawFlux = myRawSink.asFlux();
// 2. 将你的原始数据流转换为 MappedType
Flux<MappedType> myConvertedFlux = myRawFlux.map(FluxMergingExample::convertRawToMappedType);
// 3. 获取外部库的 Flux
Flux<MappedType> aFluxMap = Library.createMappingToMappedType();
// 4. 合并两个 MappedType 流
// Flux.merge 用于并行合并，元素会根据到达时间交叉输出
Flux<MappedType> combinedFlux = Flux.merge(aFluxMap, myConvertedFlux);
// 5. 订阅并处理合并后的流
combinedFlux.doOnNext(converted -> System.out.println("Received combined MappedType: " + converted.name))
.doOnComplete(() -> System.out.println("Combined Flux completed!"))
.subscribe();
// 6. 动态发射你的数据
System.out.println("Emitting custom data...");
myRawSink.tryEmitNext(new MyRawType("A")).orThrow();
Thread.sleep(100); // 稍作等待
myRawSink.tryEmitNext(new MyRawType("B")).orThrow();
Thread.sleep(300); // 稍作等待，让库的Flux也能发射一些
myRawSink.tryEmitNext(new MyRawType("C")).orThrow();
myRawSink.tryEmitComplete(); // 完成你的数据源
// 等待所有异步操作完成
Thread.sleep(1000);
}
}在 Reactor 中，有几个常用的操作符用于合并流：
- **`Flux
