// 創(chuàng)建 Mono
Mono<String> mono = Mono.just("Hello");
// 創(chuàng)建 Flux
Flux<String> flux = Flux.just("Hello", "World", "Reactor");
// 訂閱并處理結(jié)果
flux.subscribe(
    value -> System.out.println("Received: " + value),
    error -> System.err.println("Error: " + error),
    () -> System.out.println("Completed")
);

// 創(chuàng)建 Observable
Observable<String> observable = Observable.just("Hello", "World", "RxJava");
// 訂閱并處理結(jié)果
observable.subscribe(
    value -> System.out.println("Received: " + value),
    error -> System.err.println("Error: " + error),
    () -> System.out.println("Completed")
);

@RestController
public class UserController {
    @Autowired
    private UserService userService;
    @GetMapping("/users")
    public Flux<User> getUsers() {
        return userService.findAll();
    }
    @GetMapping("/users/{id}")
    public Mono<User> getUser(@PathVariable Long id) {
        return userService.findById(id);
    }
    @PostMapping("/users")
    public Mono<User> createUser(@RequestBody User user) {
        return userService.save(user);
    }
}

// 使用 limitRate 控制生產(chǎn)速度
Flux.range(1, 1000)
    .limitRate(100) // 每次請(qǐng)求 100 個(gè)元素
    .subscribe(
        value -> {
            // 處理元素
            System.out.println("Processing: " + value);
            // 模擬處理延遲
            try { Thread.sleep(10); } catch (InterruptedException e) {}
        }
    );

// 使用 onErrorReturn 處理錯(cuò)誤
Mono.just(1)
    .map(value -> {
        if (value == 1) {
            throw new RuntimeException("Error");
        }
        return value;
    })
    .onErrorReturn(0) // 錯(cuò)誤時(shí)返回默認(rèn)值
    .subscribe(System.out::println);
// 使用 onErrorResume 處理錯(cuò)誤
Mono.just(1)
    .map(value -> {
        if (value == 1) {
            throw new RuntimeException("Error");
        }
        return value;
    })
    .onErrorResume(error -> {
        // 錯(cuò)誤時(shí)返回另一個(gè) Mono
        return Mono.just(0);
    })
    .subscribe(System.out::println);

// 使用 zip 組合多個(gè) Mono
Mono<String> mono1 = Mono.just("Hello");
Mono<String> mono2 = Mono.just("World");
Mono<String> combined = Mono.zip(
    mono1, 
    mono2, 
    (s1, s2) -> s1 + " " + s2
);
combined.subscribe(System.out::println); // 輸出: Hello World
// 使用 flatMap 組合多個(gè) Flux
Flux<String> flux1 = Flux.just("A", "B");
Flux<String> flux2 = Flux.just("1", "2");
flux1.flatMap(s1 -> 
    flux2.map(s2 -> s1 + s2)
).subscribe(System.out::println); // 輸出: A1, A2, B1, B2

// 使用 parallel 并行處理
Flux.range(1, 10)
    .parallel() // 啟用并行處理
    .runOn(Schedulers.parallel()) // 指定調(diào)度器
    .map(value -> {
        // 并行處理
        System.out.println("Processing " + value + " on thread " + Thread.currentThread().getName());
        return value * 2;
    })
    .sequential() // 恢復(fù)為順序流
    .subscribe(System.out::println);

// 使用 cache 緩存結(jié)果
Mono<String> cachedMono = Mono.fromSupplier(() -> {
    System.out.println("Computing value");
    return "Hello";
}).cache();
// 第一次訂閱，會(huì)執(zhí)行計(jì)算
cachedMono.subscribe(System.out::println);
// 第二次訂閱，使用緩存的結(jié)果
cachedMono.subscribe(System.out::println);

// 使用 timeout 設(shè)置超時(shí)
Mono.just("Hello")
    .delayElement(Duration.ofSeconds(2))
    .timeout(Duration.ofSeconds(1)) // 設(shè)置 1 秒超時(shí)
    .onErrorResume(TimeoutException.class, e -> Mono.just("Timeout"))
    .subscribe(System.out::println);

@RestController
public class SensorController {
    @Autowired
    private SensorService sensorService;
    @PostMapping("/sensor/data")
    public Mono<Void> receiveData(@RequestBody Mono<SensorData> data) {
        return data.flatMap(sensorService::processData);
    }
    @GetMapping("/sensor/stats")
    public Flux<SensorStats> getStats() {
        return sensorService.getStats();
    }
}
@Service
public class SensorService {
    @Autowired
    private ReactiveMongoTemplate mongoTemplate;
    public Mono<Void> processData(SensorData data) {
        // 處理數(shù)據(jù)
        return process(data)
            // 存儲(chǔ)處理結(jié)果
            .flatMap(processedData -> 
                mongoTemplate.save(processedData)
            )
            .then();
    }
    public Flux<SensorStats> getStats() {
        // 聚合統(tǒng)計(jì)數(shù)據(jù)
        return mongoTemplate.aggregate(
            Aggregation.newAggregation(
                Aggregation.group("sensorId")
                    .avg("value").as("average")
                    .max("value").as("max")
                    .min("value").as("min")
            ),
            "sensorData",
            SensorStats.class
        );
    }
    private Mono<SensorData> process(SensorData data) {
        // 數(shù)據(jù)處理邏輯
        return Mono.just(data)
            .map(d -> {
                // 處理數(shù)據(jù)
                d.setValue(d.getValue() * 2);
                d.setProcessed(true);
                return d;
            });
    }
}