mono
  .doOnNext(日志)
  .map(转换)
  .switchIfEmpty(兜底)
  .timeout(超时)

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class RxJavaDemo {
    public static void main(String[] args) throws InterruptedException {
        // 1. 创建数据流（发布者）
        Observable<String> dataStream = Observable.create(emitter -> {
            for (int i = 1; i <= 5; i++) {
                Thread.sleep(500);
                emitter.onNext("数据->" + i);
            }
            emitter.onComplete();
        });

        // 2. 流式处理 + 订阅消费
        dataStream
                .subscribeOn(Schedulers.newThread()) // 异步线程执行
                .filter(data -> !data.equals("数据->3")) // 过滤指定数据
                .map(data -> data + "【处理完成】")     // 数据转换
                .subscribe(
                        data -> System.out.println("接收：" + data), // 正常消费
                        err -> System.err.println("异常：" + err),   // 异常捕获
                        () -> System.out.println("数据流结束")       // 流完成回调
                );

        Thread.sleep(3000);
    }
}

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class ReactorDemo {
    public static void main(String[] args) throws InterruptedException {
        // 1. 创建流式发布者
        Flux<Integer> dataStream = Flux.create(emitter -> {
            for (int i = 1; i <= 5; i++) {
                Thread.sleep(500);
                System.out.println("生产数据：" + i);
                emitter.next(i);
            }
            emitter.complete();
        });

        // 2. 链式处理 + 异步订阅
        dataStream
                .subscribeOn(Schedulers.boundedElastic()) // 异步线程池
                .filter(num -> num != 3)                  // 过滤数据
                .map(num -> num * 10)                     // 数据转换
                .doOnNext(d -> System.out.println("中间处理：" + d))
                .subscribe(
                        data -> System.out.println("最终消费：" + data),
                        err -> System.err.println("异常：" + err),
                        () -> System.out.println("数据流结束")
                );

        Thread.sleep(3500);
        System.out.println("主线程执行完毕");
    }
}

@RestController
public class OrderController {

    @GetMapping("/order/{id}")
    public Mono<Order> getOrder(@PathVariable String id) {
        // Mono：0/1 个结果 → 非阻塞查询 DB/Redis
        return orderService.findById(id)
                .doOnNext(order -> log.info("查询订单：{}", id));
    }

    @GetMapping("/orders")
    public Flux<Order> listOrders() {
        // Flux：多个结果 → 流式返回
        return orderService.findAll();
    }
}

import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

@Service
public class OrderServiceImpl implements OrderService {

    // 注入响应式 DAO（ReactiveMongo/ReactiveRedis/R2DBC）
    private final OrderReactiveDao orderDao;

    // 构造器注入（Spring推荐）
    public OrderServiceImpl(OrderReactiveDao orderDao) {
        this.orderDao = orderDao;
    }

    /**
     * 响应式根据ID查询订单
     * 重点：返回 Mono<Order>，非阻塞！
     */
    @Override
    public Mono<Order> findById(String orderId) {
        // 非阻塞查询数据库/Redis
        return orderDao.findById(orderId)
                // 数据到达后自动执行（不阻塞线程）
                .doOnNext(order -> System.out.println("查询到订单：" + orderId))
                // 空值处理
                .switchIfEmpty(Mono.error(new RuntimeException("订单不存在：" + orderId)));
    }
}

import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.r2dbc.repository.R2dbcRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Mono;

@Repository
public interface OrderDao extends R2dbcRepository<Order, String> {

    /**
     * 根据订单ID查询订单（响应式）
     */
    @Query("SELECT * FROM t_order WHERE order_id = :orderId")
    Mono<Order> findByOrderId(String orderId);
}

public Mono<UserInfo> getUserInfo(String userId) {
    // 同时调用 4 个服务，非阻塞并行执行
    Mono<User> userMono = userService.getUser(userId);
    Mono<List<Order>> orderMono = orderService.getOrders(userId);
    Mono<Coupon> couponMono = couponService.getCoupon(userId);
    Mono<Points> pointsMono = pointsService.getPoints(userId);

    // 并行聚合结果
    return Mono.zip(userMono, orderMono, couponMono, pointsMono)
            .map(tuple -> {
                // 组装返回
                return new UserInfo(tuple.getT1(), tuple.getT2(), tuple.getT3(), tuple.getT4());
            });
}

@GetMapping(value = "/monitor/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerData> monitorData() {
    // 持续推送实时数据（SSE 前端实时接收）
    return Flux.interval(Duration.ofSeconds(1))
            .map(tick -> {
                ServerData data = new ServerData();
                data.setCpu(monitorService.getCpuUsage());
                data.setMem(monitorService.getMemUsage());
                return data;
            });
}