第7章：响应式编程（Reactive Programming）

一、导读

1.1 本章学习目标

第7章介绍响应式编程——一种基于数据流和变化传播的编程范式。

本章的核心学习目标包括：

理解响应式编程的核心理念：数据流、变化传播
掌握 Observable/Observer 模式
理解背压（Backpressure）机制
学会使用 RxJS/RxJava 构建响应式应用

1.2 为什么本章如此重要

响应式编程是处理异步数据流的标准方式。从 UI 事件到 WebSocket 消息，从文件读取到 API 调用——一切都是数据流。响应式编程提供了统一的模型来处理这些流。

1.3 核心问题引导

响应式编程和回调有什么区别？

Observable 和 Promise 有什么异同？

什么是背压？为什么它很重要？

响应式编程适合什么类型的应用？

二、核心概念详解

2.1 Observable

javascript

// RxJS
import { Observable, of, from } from 'rxjs';
import { map, filter, take } from 'rxjs/operators';

// 创建 Observable
const source$ = of(1, 2, 3, 4, 5);

// 操作符链
source$.pipe(
  filter(x => x % 2 === 0),
  map(x => x * x),
  take(2)
).subscribe(x => console.log(x));    // 4, 16

// 从事件创建
const clicks$ = fromEvent(document, 'click');
clicks$.subscribe(e => console.log(e.clientX, e.clientY));

2.2 操作符

javascript

import { merge, combineLatest, concat } from 'rxjs';

// merge: 合并多个流
const merged$ = merge(stream1$, stream2$);

// combineLatest: 组合最新值
const combined$ = combineLatest([price$, quantity$]).pipe(
  map(([price, qty]) => price * qty)
);

// switchMap: 取消之前的内部流
const search$ = input$.pipe(
  debounceTime(300),
  switchMap(query => searchApi(query))
);

2.3 背压

javascript

import { interval } from 'rxjs';
import { bufferTime } from 'rxjs/operators';

// 快速生产者，慢速消费者
const fast$ = interval(10);    // 每10ms一个值

// 使用 bufferTime 批量处理
fast$.pipe(
  bufferTime(1000)    // 每1秒收集一批
).subscribe(batch => {
  console.log(`Processing ${batch.length} items`);
});

三、实战案例

3.1 案例一：搜索自动补全

javascript

import { fromEvent } from 'rxjs';
import { map, debounceTime, distinctUntilChanged, switchMap, catchError } from 'rxjs/operators';

const input = document.getElementById('search');

const search$ = fromEvent(input, 'input').pipe(
  map(e => e.target.value),
  debounceTime(300),
  distinctUntilChanged(),
  switchMap(query => 
    fetch(`/api/search?q=${query}`)
      .then(res => res.json())
      .catch(() => [])
  )
);

search$.subscribe(results => {
  renderResults(results);
});

3.2 案例二：实时数据仪表盘

javascript

import { WebSocketSubject } from 'rxjs/webSocket';
import { groupBy, mergeMap, scan } from 'rxjs/operators';

const ws$ = new WebSocketSubject('ws://localhost:8080/data');

const dashboard$ = ws$.pipe(
  groupBy(data => data.type),
  mergeMap(group$ => 
    group$.pipe(
      scan((acc, val) => [...acc.slice(-9), val], []),
      map(values => ({ type: group$.key, values }))
    )
  )
);

dashboard$.subscribe(({ type, values }) => {
  updateChart(type, values);
});

四、本章小结

响应式编程：基于数据流和变化传播的编程范式。

Observable：表示异步数据流的核心抽象。

操作符：map、filter、switchMap 等组合数据流。

背压：处理生产者快于消费者的策略。

五、思考与练习

响应式编程和回调地狱有什么关系？它如何解决异步编程的复杂性？

switchMap 和 mergeMap 有什么区别？各适用于什么场景？

用响应式编程实现一个简单的拖拽功能。