microwind/ai-skills/reactive-programming

Observable:   --1--2--3--4--5--|
                 │
filter(x>2):  --------3--4--5--|
                 │
map(x*10):    --------30-40-50-|
                 │
Observer:     --------30-40-50-|
 
--  时间轴
|   完成
X   错误

// ✅ 响应式处理异步数据流
Flux<Order> orders = orderRepository.findAll();  // 响应式数据源
 
orders
    .filter(order -> order.getStatus() == OrderStatus.PENDING)
    .map(order -> {
        order.process();
        return order;
    })
    .flatMap(order -> paymentService.charge(order))  // 异步调用
    .doOnNext(order -> log.info("处理完成: {}", order.getId()))
    .doOnError(error -> log.error("处理失败", error))
    .subscribe();
 
// 组合多个数据流
Mono<User> user = userService.findById(userId);
Mono<List<Order>> orders = orderService.findByUser(userId);
 
Mono.zip(user, orders)
    .map(tuple -> new UserDashboard(tuple.getT1(), tuple.getT2()))
    .subscribe(dashboard -> renderPage(dashboard));

import { fromEvent, interval, merge } from 'rxjs';
import { map, filter, debounceTime, switchMap, takeUntil } from 'rxjs/operators';
// 搜索框：防抖 + 取消前一个请求
const search$ = fromEvent<InputEvent>(searchInput, 'input').pipe(
    map(event => (event.target as HTMLInputElement).value),
    filter(query => query.length >= 2),
    debounceTime(300),                              // 300ms 防抖
    switchMap(query => fetch(`/api/search?q=${query}`)  // 取消前一个请求
        .then(r => r.json()))
);
search$.subscribe(results => renderResults(results));
// 实时数据流
const stockPrice$ = new WebSocket('ws://prices');
const ticker$ = fromEvent(stockPrice$, 'message').pipe(
    map(msg => JSON.parse(msg.data)),
    filter(price => price.symbol === 'AAPL'),
    map(price => price.value)
);
ticker$.subscribe(price => updateChart(price));

import rx
from rx import operators as ops
# 数据处理管道
source = rx.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
source.pipe(
    ops.filter(lambda x: x % 2 == 0),
    ops.map(lambda x: x ** 2),
    ops.reduce(lambda acc, x: acc + x)
).subscribe(
    on_next=lambda result: print(f"结果: {result}"),
    on_error=lambda e: print(f"错误: {e}"),
    on_completed=lambda: print("完成")
)

// 生产者速度 > 消费者速度时的处理策略
 
Flux.range(1, 1000000)
    .onBackpressureBuffer(1000)    // 缓冲区
    // .onBackpressureDrop()       // 丢弃多余的
    // .onBackpressureLatest()     // 只保留最新的
    .subscribe(new BaseSubscriber<Integer>() {
        @Override
        protected void hookOnSubscribe(Subscription s) {
            request(10);  // 一次只请求10个
        }
 
        @Override
        protected void hookOnNext(Integer value) {
            process(value);
            request(10);  // 处理完再请求
        }
    });

✅ 非常适合：
- UI 事件处理（点击、输入、滚动）
- 实时数据流（股票行情、消息推送）
- 异步 API 组合（多个服务并行调用）
- WebSocket 数据处理
- 搜索自动补全（防抖+取消）
 
❌ 不太适合：
- 简单的同步 CRUD
- 计算密集型任务
- 团队不熟悉响应式概念