WebFlux 响应式编程实战：Reactor 内核、响应式数据访问与选型

// 内部：每个操作符生成新的 Flux，持引用上游
// 订阅时：subscribe → 逐层调用 onSubscribe，建立反向的订阅关系
// 数据时：上游 onNext 逐层向下游传递，每个操作符处理后再转发
// 背压时：下游 request(n) 逐层向上游传递，上游按需生产

Flux.range(1, 100)
    .parallel(8)                            // 并行度 8
    .runOn(Schedulers.parallel())           // 计算密集在 parallel
    .map(this::cpuBoundTask)
    .flatMap(item -> blockingIo(item)       // 阻塞调用必须隔离！
        .subscribeOn(Schedulers.boundedElastic()))  // 在弹性线程池执行
    .subscribe();

// Reactor 内部的核心契约：
// Publisher.subscribe(Subscriber)
//   → Subscriber.onSubscribe(Subscription)
//   → Subscription.request(n)  ← 下游向上游申请 N 个元素
//   → 上游每生产一个元素调用 Subscriber.onNext(t)
//   → 数据流结束 onComplete() / 异常 onError()

// 自定义一个支持背压的 Publisher：
Flux.create(sink -> {
    for (int i = 0; i < 100_000; i++) {
        sink.next(i);                       // 生产
    }
    sink.complete();
}, FluxSink.OverflowStrategy.BUFFER)        // 背压溢出策略
    .limitRate(500);                        // 上游每批最多 500，防止内存爆

// 多订阅者共享同一数据流（热流场景）
Sinks.Many<Message> sink = Sinks.many().multicast().onBackpressureBuffer();
sink.tryEmitNext(new Message("hello"));     // 生产端

Flux<Message> shared = sink.asFlux()
    .publish()                              // 转热流
    .autoConnect(1);                        // 第一个订阅才启动

// ❌ 反模式：在 WebFlux 中阻塞
@GetMapping("/bad")
public String bad() {
    return jdbcTemplate.queryForObject(...);   // 阻塞 EventLoop！
}

// ❌ 反模式：把 Mono 阻塞住
@GetMapping("/bad2")
public String bad2() {
    Mono<String> mono = monoService.call();
    return mono.block();                       // block() 阻塞当前线程
}

// ✅ 正确：全程非阻塞
@GetMapping("/good")
public Mono<Order> good() {
    return orderRepo.findById(id)              // R2DBC 非阻塞
        .map(order -> enrich(order))           // CPU 操作留在 EventLoop
        .timeout(Duration.ofSeconds(2));
}

// ✅ 若必须调用阻塞 SDK，隔离到 boundedElastic
@GetMapping("/isolated")
public Mono<String> isolated() {
    return Mono.fromCallable(() -> legacyClient.syncCall())
        .subscribeOn(Schedulers.boundedElastic());
}

@GetMapping("/orders")
public Mono<List<Order>> getOrders() {
    return orderService.findRecent()
        .timeout(Duration.ofMillis(500))                       // 整体超时
        .onErrorResume(TimeoutException.class,
            e -> Mono.just(Collections.emptyList()))           // 超时降级为空列表
        .retryWhen(Retry.backoff(3, Duration.ofMillis(100))    // 指数退避重试
            .maxBackoff(Duration.ofSeconds(2)));
}

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>
<dependency>
    <groupId>io.r2dbc</groupId>
    <artifactId>r2dbc-mysql</artifactId>      <!-- 或 r2dbc-postgresql -->
</dependency>

spring:
  r2dbc:
    url: r2dbc:mysql://localhost:3306/order_db
    username: root
    password: secret
    pool:
      enabled: true
      initial-size: 4
      max-size: 32
      max-idle-time: 30m

@Service
public class OrderTxService {
    @Transactional
    public Mono<Void> createOrderWithItems(OrderRequest req) {
        return orderRepo.save(req.toOrder())
            .flatMap(order -> itemRepo.saveAll(req.getItems(order.getId()))
                .then());
    }
}
// R2DBC 事务与响应式结合：ReactiveTransactionManager 管理
// @Transactional 作用于返回 Mono/Flux 的方法，提交在流完成时进行

public interface OrderReactiveRepository extends ReactiveMongoRepository<Order, String> {
    Flux<Order> findByUserId(String userId);
    Mono<Order> findFirstByStatusOrderByCreatedAtDesc(OrderStatus status);

    @Aggregation(pipeline = {
        "{ '$match': { 'status': 'PAID' } }",
        "{ '$group': { '_id': '$userId', 'total': { '$sum': '$amount' } } }"
    })
    Flux<UserSpending> aggregateSpending();
}

@RestController
public class OrderController {
    @GetMapping("/reactive/orders/{userId}")
    public Flux<Order> list(@PathVariable String userId) {
        return repo.findByUserId(userId)
            .onBackpressureBuffer();           // 下游消费慢时缓冲，防止无限生产
    }
}

@Query("SELECT * FROM orders WHERE user_id = :uid AND amount > :min ORDER BY created_at DESC")
Flux<Order> findBigOrders(@Param("uid") String uid, @Param("min") BigDecimal min);

@Query("SELECT * FROM orders WHERE user_id = :uid AND id < :cursor ORDER BY id DESC LIMIT 20")
Flux<Order> findByCursor(@Param("uid") String uid, @Param("cursor") String cursor);

// 显式控制下游消费速率
Flux<Order> orderStream = orderService.streamAll();
orderStream.subscribe(new BaseSubscriber<Order>() {
    @Override
    protected void hookOnSubscribe(Subscription subscription) {
        subscription.request(10);            // 第一批只要 10 个
    }
    @Override
    protected void hookOnNext(Order order) {
        process(order);
        request(1);                          // 每处理一个再要一个（类似限速）
    }
});

Flux.interval(Duration.ofMillis(10))
    .onBackpressureDrop(ts -> log.warn("dropped {}", ts))   // 丢弃并记录
    .concatMap(tick -> expensiveOp(tick), 1)                // 串行消费，天然背压

Flux.range(1, 1_000_000)
    .limitRate(256)                        // 每次 request(256)，吞吐与内存平衡
    .bufferTimeout(500, Duration.ofMillis(50))  // 批量 + 时间窗，降低下游调用频率
    .concatMap(batch -> saveBatch(batch))  // 批量写库

// Spring MVC + 虚拟线程：Tomcat 可以配置为每请求一个虚拟线程
spring:
  threads:
    virtual:
      enabled: true

// 结论：若目标只是「高并发 + 阻塞 JDBC + 同步代码」，虚拟线程 + MVC 更务实
// WebFlux 的独特价值在于：真正的事件驱动 + 背压 + 海量长连接

// MVC Controller 内调用 WebClient（非阻塞 HTTP 客户端）
@GetMapping("/mvc")
public ResponseEntity<Result> callRemote() {
    Mono<RemoteResp> mono = webClient.get()
        .uri("http://user-service/users/{id}", id)
        .retrieve()
        .bodyToMono(RemoteResp.class)
        .timeout(Duration.ofMillis(800));
    return ResponseEntity.ok(mono.block());    // MVC 栈允许 block，线程模型支持
}
// 优势：复用 WebClient 的连接池、超时重试、非阻塞传输
// 劣势：block() 会阻塞 MVC 线程，但比 HttpClient 阻塞式更好

// 混合：入口 MVC，数据源阻塞，用 reactive 客户端优化上游调用
// 或：入口 WebFlux，数据源是 Mongo（响应式），部分外部依赖用 boundedElastic 隔离