Java中Flux类响应式编程的核心组件详解

// 1. 从固定值创建
Flux<String> flux1 = Flux.just("A", "B", "C");
// 2. 从数组创建
Flux<String> flux2 = Flux.fromArray(new String[]{"A", "B", "C"});
// 3. 从Iterable创建
Flux<String> flux3 = Flux.fromIterable(Arrays.asList("A", "B", "C"));
// 4. 从Stream创建
Flux<String> flux4 = Flux.fromStream(Stream.of("A", "B", "C"));
// 5. 创建一个范围数字Flux
Flux<Integer> flux5 = Flux.range(1, 5); // 1, 2, 3, 4, 5
// 6. 创建一个空的Flux
Flux<String> flux6 = Flux.empty();
// 7. 创建一个错误Flux
Flux<String> flux7 = Flux.error(new RuntimeException("Error occurred"));
// 8. 创建一个无限序列
Flux<Long> flux8 = Flux.interval(Duration.ofSeconds(1)); // 每秒发射一个递增的数字

// 使用generate创建有状态的Flux
Flux<String> statefulFlux = Flux.generate(
    () -> 0, // 初始状态
    (state, sink) -> {
        sink.next("3 x " + state + " = " + 3*state);
        if (state == 10) sink.complete();
        return state + 1;
    }
);
// 使用create创建更复杂的Flux
Flux<String> complexFlux = Flux.create(emitter -> {
    // 可以异步地发射多个元素
    emitter.next("First");
    emitter.next("Second");
    emitter.complete();
});

Flux<Integer> numbers = Flux.range(1, 5);
// map - 转换元素
Flux<Integer> squared = numbers.map(n -> n * n);
// flatMap - 异步转换元素为Publisher
Flux<Integer> flatMapped = numbers.flatMap(n -> Mono.just(n * 2));
// concatMap - 保持顺序的flatMap
Flux<Integer> concatMapped = numbers.concatMap(n -> Mono.just(n * 2));
// buffer - 缓冲元素
Flux<List<Integer>> buffered = numbers.buffer(2); // [[1,2], [3,4], [5]]

// filter - 过滤元素
Flux<Integer> evens = numbers.filter(n -> n % 2 == 0);
// take - 取前N个元素
Flux<Integer> firstThree = numbers.take(3);
// skip - 跳过前N个元素
Flux<Integer> afterTwo = numbers.skip(2);
// distinct - 去重
Flux<Integer> distinct = Flux.just(1, 2, 2, 3, 3, 3).distinct();

Flux<String> fluxA = Flux.just("A", "B", "C");
Flux<String> fluxB = Flux.just("D", "E", "F");
// merge - 合并多个Flux，不保证顺序
Flux<String> merged = Flux.merge(fluxA, fluxB);
// concat - 顺序连接多个Flux
Flux<String> concatenated = Flux.concat(fluxA, fluxB);
// zip - 将多个Flux的元素配对组合
Flux<String> zipped = Flux.zip(fluxA, fluxB, (a, b) -> a + b); // ["AD", "BE", "CF"]
// combineLatest - 每当任一Flux发射时组合最新值
Flux<String> combined = Flux.combineLatest(
    fluxA, 
    fluxB, 
    (a, b) -> a + b
);

Flux<String> errorFlux = Flux.error(new RuntimeException("Error"));
// onErrorReturn - 发生错误时返回默认值
Flux<String> withDefault = errorFlux.onErrorReturn("Default");
// onErrorResume - 发生错误时切换到一个备用的Publisher
Flux<String> withFallback = errorFlux.onErrorResume(e -> Flux.just("Fallback"));
// retry - 重试
Flux<String> retried = errorFlux.retry(3); // 最多重试3次
// retryWhen - 条件重试
Flux<String> retriedWhen = errorFlux.retryWhen(
    Retry.backoff(3, Duration.ofSeconds(1))
);

Flux<String> flux = Flux.just("Hello", "World");
// 1. 最简单的订阅
flux.subscribe();
// 2. 带消费者回调的订阅
flux.subscribe(
    value -> System.out.println("Received: " + value), // onNext
    error -> System.err.println("Error: " + error),    // onError
    () -> System.out.println("Completed")             // onComplete
);
// 3. 带Subscription控制的订阅
flux.subscribe(
    value -> System.out.println("Received: " + value),
    error -> System.err.println("Error: " + error),
    () -> System.out.println("Completed"),
    subscription -> {
        subscription.request(1); // 请求1个元素
        // 可以在这里保存subscription以便后续控制
    }
);

// 转换为Iterable(阻塞)
Iterable<String> iterable = flux.toIterable();
// 收集到List(阻塞)
List<String> list = flux.collectList().block();
// 收集到Mono(非阻塞)
Mono<List<String>> monoList = flux.collectList();

@RestController
public class UserController {
    @GetMapping("/users")
    public Flux<User> getUsers() {
        return userRepository.findAll(); // 返回Flux<User>
    }
    @GetMapping("/users/{id}")
    public Mono<User> getUser(@PathVariable String id) {
        return userRepository.findById(id); // 返回Mono<User>
    }
}

Flux.fromIterable(dataSource)
    .filter(data -> data.isValid())
    .map(data -> transform(data))
    .buffer(100)
    .flatMap(batch -> saveToDatabase(batch))
    .onErrorResume(e -> logAndContinue(e))
    .subscribe();

// 模拟事件源
Flux<Event> eventStream = Flux.interval(Duration.ofMillis(100))
    .map(tick -> generateRandomEvent());
// 处理事件流
eventStream
    .groupBy(Event::getType) // 按类型分组
    .flatMap(groupedFlux -> 
        groupedFlux
            .window(Duration.ofSeconds(1))
            .flatMap(window -> 
                window.reduce(new EventAccumulator(), this::accumulate)
            )
    )
    .subscribe(accumulator -> System.out.println("Processed: " + accumulator));

// 使用R2DBC进行响应式数据库访问
Flux<User> activeUsers = databaseClient
    .sql("SELECT * FROM users WHERE status = :status")
    .bind("status", "ACTIVE")
    .map((row, metadata) -> new User(
        row.get("id", String.class),
        row.get("name", String.class)
    ))
    .all();

// 冷发布示例
Flux<Integer> cold = Flux.range(1, 3); // 每次订阅都会重新发射1,2,3
// 转换为热发布
ConnectableFlux<Integer> hot = cold.publish();
hot.connect(); // 开始发射数据，所有订阅者共享相同的数据流

Flux.range(1, 100)
    .onBackpressureBuffer(10) // 缓冲区大小为10
    .subscribe(new BaseSubscriber<Integer>() {
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            request(1); // 每次只请求1个元素
        }
        @Override
        protected void hookOnNext(Integer value) {
            System.out.println("Received: " + value);
            // 处理完后再请求下一个
            request(1);
        }
    });

Flux.range(1, 10)
    .parallel() // 并行处理
    .runOn(Schedulers.parallel()) // 在并行线程池上运行
    .map(i -> i * 2)
    .sequential() // 切换回单线程
    .subscribeOn(Schedulers.single()) // 订阅在单一线程
    .publishOn(Schedulers.elastic()) // 后续操作在弹性线程池
    .subscribe();

@Test
public void testFlux() {
    Flux<String> flux = Flux.just("foo", "bar");
    // 使用StepVerifier测试
    StepVerifier.create(flux)
        .expectNext("foo")
        .expectNext("bar")
        .expectComplete()
        .verify();
    // 测试错误情况
    Flux<String> errorFlux = Flux.error(new RuntimeException());
    StepVerifier.create(errorFlux)
        .expectError(RuntimeException.class)
        .verify();
}