Java 反应式编程构建响应式系统的实践案例

// 创建 Mono
Mono<String> mono = Mono.just("Hello");
// 创建 Flux
Flux<String> flux = Flux.just("Hello", "World", "Reactor");
// 订阅并处理结果
flux.subscribe(
    value -> System.out.println("Received: " + value),
    error -> System.err.println("Error: " + error),
    () -> System.out.println("Completed")
);

// 创建 Observable
Observable<String> observable = Observable.just("Hello", "World", "RxJava");
// 订阅并处理结果
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 控制生产速度
Flux.range(1, 1000)
    .limitRate(100) // 每次请求 100 个元素
    .subscribe(
        value -> {
            // 处理元素
            System.out.println("Processing: " + value);
            // 模拟处理延迟
            try { Thread.sleep(10); } catch (InterruptedException e) {}
        }
    );

// 使用 onErrorReturn 处理错误
Mono.just(1)
    .map(value -> {
        if (value == 1) {
            throw new RuntimeException("Error");
        }
        return value;
    })
    .onErrorReturn(0) // 错误时返回默认值
    .subscribe(System.out::println);
// 使用 onErrorResume 处理错误
Mono.just(1)
    .map(value -> {
        if (value == 1) {
            throw new RuntimeException("Error");
        }
        return value;
    })
    .onErrorResume(error -> {
        // 错误时返回另一个 Mono
        return Mono.just(0);
    })
    .subscribe(System.out::println);

// 使用 zip 组合多个 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 组合多个 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()) // 指定调度器
    .map(value -> {
        // 并行处理
        System.out.println("Processing " + value + " on thread " + Thread.currentThread().getName());
        return value * 2;
    })
    .sequential() // 恢复为顺序流
    .subscribe(System.out::println);

// 使用 cache 缓存结果
Mono<String> cachedMono = Mono.fromSupplier(() -> {
    System.out.println("Computing value");
    return "Hello";
}).cache();
// 第一次订阅，会执行计算
cachedMono.subscribe(System.out::println);
// 第二次订阅，使用缓存的结果
cachedMono.subscribe(System.out::println);

// 使用 timeout 设置超时
Mono.just("Hello")
    .delayElement(Duration.ofSeconds(2))
    .timeout(Duration.ofSeconds(1)) // 设置 1 秒超时
    .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) {
        // 处理数据
        return process(data)
            // 存储处理结果
            .flatMap(processedData -> 
                mongoTemplate.save(processedData)
            )
            .then();
    }
    public Flux<SensorStats> getStats() {
        // 聚合统计数据
        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) {
        // 数据处理逻辑
        return Mono.just(data)
            .map(d -> {
                // 处理数据
                d.setValue(d.getValue() * 2);
                d.setProcessed(true);
                return d;
            });
    }
}