// 1. 從固定值創(chuàng)建
Flux<String> flux1 = Flux.just("A", "B", "C");
// 2. 從數(shù)組創(chuàng)建
Flux<String> flux2 = Flux.fromArray(new String[]{"A", "B", "C"});
// 3. 從Iterable創(chuàng)建
Flux<String> flux3 = Flux.fromIterable(Arrays.asList("A", "B", "C"));
// 4. 從Stream創(chuàng)建
Flux<String> flux4 = Flux.fromStream(Stream.of("A", "B", "C"));
// 5. 創(chuàng)建一個(gè)范圍數(shù)字Flux
Flux<Integer> flux5 = Flux.range(1, 5); // 1, 2, 3, 4, 5
// 6. 創(chuàng)建一個(gè)空的Flux
Flux<String> flux6 = Flux.empty();
// 7. 創(chuàng)建一個(gè)錯(cuò)誤Flux
Flux<String> flux7 = Flux.error(new RuntimeException("Error occurred"));
// 8. 創(chuàng)建一個(gè)無(wú)限序列
Flux<Long> flux8 = Flux.interval(Duration.ofSeconds(1)); // 每秒發(fā)射一個(gè)遞增的數(shù)字

// 使用generate創(chuàng)建有狀態(tài)的Flux
Flux<String> statefulFlux = Flux.generate(
    () -> 0, // 初始狀態(tài)
    (state, sink) -> {
        sink.next("3 x " + state + " = " + 3*state);
        if (state == 10) sink.complete();
        return state + 1;
    }
);
// 使用create創(chuàng)建更復(fù)雜的Flux
Flux<String> complexFlux = Flux.create(emitter -> {
    // 可以異步地發(fā)射多個(gè)元素
    emitter.next("First");
    emitter.next("Second");
    emitter.complete();
});

Flux<Integer> numbers = Flux.range(1, 5);
// map - 轉(zhuǎn)換元素
Flux<Integer> squared = numbers.map(n -> n * n);
// flatMap - 異步轉(zhuǎn)換元素為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 - 過(guò)濾元素
Flux<Integer> evens = numbers.filter(n -> n % 2 == 0);
// take - 取前N個(gè)元素
Flux<Integer> firstThree = numbers.take(3);
// skip - 跳過(guò)前N個(gè)元素
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 - 合并多個(gè)Flux，不保證順序
Flux<String> merged = Flux.merge(fluxA, fluxB);
// concat - 順序連接多個(gè)Flux
Flux<String> concatenated = Flux.concat(fluxA, fluxB);
// zip - 將多個(gè)Flux的元素配對(duì)組合
Flux<String> zipped = Flux.zip(fluxA, fluxB, (a, b) -> a + b); // ["AD", "BE", "CF"]
// combineLatest - 每當(dāng)任一Flux發(fā)射時(shí)組合最新值
Flux<String> combined = Flux.combineLatest(
    fluxA, 
    fluxB, 
    (a, b) -> a + b
);

Flux<String> errorFlux = Flux.error(new RuntimeException("Error"));
// onErrorReturn - 發(fā)生錯(cuò)誤時(shí)返回默認(rèn)值
Flux<String> withDefault = errorFlux.onErrorReturn("Default");
// onErrorResume - 發(fā)生錯(cuò)誤時(shí)切換到一個(gè)備用的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. 最簡(jiǎn)單的訂閱
flux.subscribe();
// 2. 帶消費(fèi)者回調(diào)的訂閱
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); // 請(qǐng)求1個(gè)元素
        // 可以在這里保存subscription以便后續(xù)控制
    }
);

// 轉(zhuǎn)換為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進(jìn)行響應(yīng)式數(shù)據(jù)庫(kù)訪問(wèn)
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();

// 冷發(fā)布示例
Flux<Integer> cold = Flux.range(1, 3); // 每次訂閱都會(huì)重新發(fā)射1,2,3
// 轉(zhuǎn)換為熱發(fā)布
ConnectableFlux<Integer> hot = cold.publish();
hot.connect(); // 開(kāi)始發(fā)射數(shù)據(jù)，所有訂閱者共享相同的數(shù)據(jù)流

Flux.range(1, 100)
    .onBackpressureBuffer(10) // 緩沖區(qū)大小為10
    .subscribe(new BaseSubscriber<Integer>() {
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            request(1); // 每次只請(qǐng)求1個(gè)元素
        }
        @Override
        protected void hookOnNext(Integer value) {
            System.out.println("Received: " + value);
            // 處理完后再請(qǐng)求下一個(gè)
            request(1);
        }
    });

Flux.range(1, 10)
    .parallel() // 并行處理
    .runOn(Schedulers.parallel()) // 在并行線程池上運(yùn)行
    .map(i -> i * 2)
    .sequential() // 切換回單線程
    .subscribeOn(Schedulers.single()) // 訂閱在單一線程
    .publishOn(Schedulers.elastic()) // 后續(xù)操作在彈性線程池
    .subscribe();

@Test
public void testFlux() {
    Flux<String> flux = Flux.just("foo", "bar");
    // 使用StepVerifier測(cè)試
    StepVerifier.create(flux)
        .expectNext("foo")
        .expectNext("bar")
        .expectComplete()
        .verify();
    // 測(cè)試錯(cuò)誤情況
    Flux<String> errorFlux = Flux.error(new RuntimeException());
    StepVerifier.create(errorFlux)
        .expectError(RuntimeException.class)
        .verify();
}