RxJS 深入浅出:从 Promise 用户到响应式编程,你需要补的全部认知

import { fromEvent } from 'rxjs'
import { debounceTime, map, switchMap } from 'rxjs/operators'
import { ajax } from 'rxjs/ajax'

fromEvent(input, 'input').pipe(
  debounceTime(300),
  map((e: any) => e.target.value),
  switchMap((q) => ajax.getJSON(`/api/search?q=${q}`)),
).subscribe((res) => render(res))

import { Observable } from 'rxjs'

const obs$ = new Observable<number>((subscriber) => {
  subscriber.next(1)        // 产出值 1
  subscriber.next(2)        // 产出值 2
  setTimeout(() => {
    subscriber.next(3)      // 1 秒后产出值 3
    subscriber.complete()   // 流结束
  }, 1000)
})

console.log('before')   // 立即打印
obs$.subscribe((v) => console.log(v))  // 这里才开始 next(1), next(2)...
console.log('after')    // 这条会先于 next(3) 打印

obs$.subscribe({
  next: (v) => console.log('值:', v),
  error: (e) => console.error('出错:', e),
  complete: () => console.log('结束'),
})

obs$.subscribe((v) => console.log(v))

const sub = obs$.subscribe((v) => console.log(v))

// 5 秒后取消订阅,后面 next 出来的值都不会再触发回调
setTimeout(() => sub.unsubscribe(), 5000)

import { of } from 'rxjs'
import { map, filter } from 'rxjs/operators'

of(1, 2, 3, 4, 5).pipe(
  filter((n) => n % 2 === 0),  // 只保留偶数
  map((n) => n * 10),           // 每个乘以 10
).subscribe((v) => console.log(v))  // 输出 20, 40

import { of, from, interval, timer, fromEvent, EMPTY, throwError } from 'rxjs'

of(1, 2, 3)                              // 同步发出 1, 2, 3 立即 complete
from([1, 2, 3])                          // 跟 of 类似,可以接数组/Promise/Iterable
from(fetch('/api/users'))                // Promise → Observable
interval(1000)                           // 每 1 秒发出 0, 1, 2, 3...(无限流)
timer(2000, 500)                         // 2 秒后开始,每 500ms 发一个
fromEvent(button, 'click')               // DOM 事件流
EMPTY                                     // 立即 complete 的空流
throwError(() => new Error('boom'))      // 立即 error

from(fetch('/api/posts').then((r) => r.json()))
  .pipe(
    catchError((err) => of({ error: err.message })),
  )
  .subscribe((data) => console.log(data))

import { map, scan, pluck } from 'rxjs/operators'

// map:一对一变换
of(1, 2, 3).pipe(map((n) => n * 2))  // 2, 4, 6

// scan:类似数组的 reduce,但每一步都吐出当前累计值(做累加/状态管理用)
of(1, 2, 3, 4).pipe(scan((acc, cur) => acc + cur, 0))  // 1, 3, 6, 10

// pluck(deprecated, 用 map 代替):从对象里取字段
of({ user: { name: 'Tom' } }).pipe(map((x) => x.user.name))  // 'Tom'

const action$ = new Subject<Action>()
const state$ = action$.pipe(
  scan((state, action) => reducer(state, action), initialState),
)

import { filter, take, skip, debounceTime, throttleTime, distinctUntilChanged } from 'rxjs/operators'

filter((n) => n > 10)                  // 只留大于 10 的
take(3)                                 // 只取前 3 个就 complete
skip(2)                                 // 跳过前 2 个
debounceTime(300)                       // 防抖:停顿 300ms 才发(最新的)
throttleTime(300)                       // 节流:每 300ms 最多发 1 个(第一个)
distinctUntilChanged()                  // 跟上一个相同就不发(去连续重复)

fromEvent(window, 'scroll').pipe(
  throttleTime(100),  // 每 100ms 最多处理 1 次
).subscribe(() => updateProgressBar())

import { fromEvent, interval } from 'rxjs'
import { switchMap, mergeMap, concatMap, exhaustMap, take } from 'rxjs/operators'

const click$ = fromEvent(button, 'click')

// switchMap:每次点击都取消上次的,只跑最新一次
click$.pipe(switchMap(() => interval(1000).pipe(take(3))))

// mergeMap:每次点击都启动一个新的,所有都并行跑
click$.pipe(mergeMap(() => interval(1000).pipe(take(3))))

// concatMap:每次点击都排队,前一个跑完才轮到下一个
click$.pipe(concatMap(() => interval(1000).pipe(take(3))))

// exhaustMap:正在跑的时候点击没反应,跑完才接受下一次
click$.pipe(exhaustMap(() => interval(1000).pipe(take(3))))

import { combineLatest, merge, zip, forkJoin, race, withLatestFrom } from 'rxjs'

// combineLatest:任意一个流出新值,合并所有流的"最新值"发出
combineLatest([user$, settings$, theme$])
  .subscribe(([user, settings, theme]) => render(user, settings, theme))

// merge:把多个流的值平铺成一个流,谁来值就发谁
merge(click$, keydown$).subscribe((evt) => handle(evt))

// zip:严格按位置配对,每个流出一个值才合并一次(像拉链)
zip(name$, age$).subscribe(([name, age]) => console.log(name, age))

// forkJoin:像 Promise.all,等所有流都 complete,取每个的最后一个值
forkJoin([fetchUser(), fetchPosts(), fetchSettings()])
  .subscribe(([user, posts, settings]) => initialize(user, posts, settings))

// withLatestFrom:左边流触发时,带上右边流的最新值
click$.pipe(withLatestFrom(user$))
  .subscribe(([clickEvt, currentUser]) => log(clickEvt, currentUser))

import { catchError, retry, retryWhen, throwError } from 'rxjs/operators'
import { of, timer } from 'rxjs'

ajax.getJSON('/api/data').pipe(
  retry(3),                                        // 失败重试 3 次
  catchError((err) => of({ error: err.message })), // 兜底返回默认值
)

// 指数退避重试:1s, 2s, 4s
ajax.getJSON('/api/data').pipe(
  retryWhen((errors) =>
    errors.pipe(
      scan((retries, _) => {
        if (retries >= 3) throw new Error('重试 3 次仍失败')
        return retries + 1
      }, 0),
      delayWhen((retries) => timer(2 ** retries * 1000)),
    ),
  ),
)

import { Subject } from 'rxjs'

const subject$ = new Subject<number>()

subject$.subscribe((v) => console.log('订阅者 A:', v))

subject$.next(1)  // 输出:订阅者 A: 1
subject$.next(2)  // 输出:订阅者 A: 2

subject$.subscribe((v) => console.log('订阅者 B:', v))

subject$.next(3)  // 输出:订阅者 A: 3 和 订阅者 B: 3
                  // ⚠️ 订阅者 B 拿不到之前的 1 和 2!

const currentUser$ = new BehaviorSubject<User | null>(null)

// 任何地方
currentUser$.subscribe((user) => console.log('当前用户:', user))

// 登录成功后
currentUser$.next({ id: 1, name: 'Tom' })

// 任何时候想拿当前值
const user = currentUser$.value

const cold$ = ajax.getJSON('/api/posts')

cold$.subscribe(...)  // 发起第 1 次请求
cold$.subscribe(...)  // 发起第 2 次请求(共 2 次!)

const click$ = fromEvent(button, 'click')  // 一创建就在监听,不订阅也在跑

const shared$ = ajax.getJSON('/api/posts').pipe(shareReplay(1))

shared$.subscribe(...)  // 发起请求,缓存结果
shared$.subscribe(...)  // 直接拿缓存,不再请求

fromEvent(searchInput, 'input').pipe(
  map((e: any) => e.target.value.trim()),
  debounceTime(300),
  distinctUntilChanged(),        // 输入没变就不发
  filter((q) => q.length >= 2),  // 至少 2 个字才搜
  switchMap((q) =>
    ajax.getJSON(`/api/search?q=${q}`).pipe(
      catchError(() => of({ results: [] })),
    ),
  ),
).subscribe((res) => renderSuggestions(res))

import { timer } from 'rxjs'
import { switchMap, takeWhile } from 'rxjs/operators'

timer(0, 3000).pipe(                          // 立即开始,然后每 3 秒
  switchMap(() => ajax.getJSON('/api/job-status')),
  takeWhile((res: any) => res.status !== 'done', true),  // done 后停止
).subscribe((res) => updateProgress(res))

const upload$ = new Observable<number>((subscriber) => {
  const xhr = new XMLHttpRequest()
  xhr.upload.onprogress = (e) => subscriber.next((e.loaded / e.total) * 100)
  xhr.onload = () => subscriber.complete()
  xhr.onerror = () => subscriber.error('上传失败')
  xhr.open('POST', '/api/upload')
  xhr.send(file)
  return () => xhr.abort()  // 取消订阅时自动 abort!
})

const sub = upload$.subscribe({
  next: (pct) => console.log(`${pct.toFixed(1)}%`),
  complete: () => console.log('完成'),
})

cancelButton.addEventListener('click', () => sub.unsubscribe())  // 触发 xhr.abort()

from(urls).pipe(
  mergeMap((url) => ajax.getJSON(url), 5),  // 并发上限 5
).subscribe((res) => collect(res))

// ❌ 坑爹写法
onMounted(() => {
  interval(1000).subscribe((n) => console.log(n))
  // 组件销毁后这个 interval 还在跑!
})

let sub: Subscription
onMounted(() => {
  sub = interval(1000).subscribe(...)
})
onUnmounted(() => sub?.unsubscribe())

const destroy$ = new Subject<void>()

interval(1000).pipe(
  takeUntil(destroy$),
).subscribe(...)

ajax.getJSON('/api/x').pipe(
  takeUntil(destroy$),
).subscribe(...)

onUnmounted(() => {
  destroy$.next()
  destroy$.complete()
})

function useSubscription() {
  const subs: Subscription[] = []
  onUnmounted(() => subs.forEach((s) => s.unsubscribe()))
  return {
    add(s: Subscription) { subs.push(s); return s }
  }
}

const { add } = useSubscription()
add(interval(1000).subscribe(...))

const user = await firstValueFrom(getUser$('123'))

from(fetch('/api/user').then((r) => r.json())).pipe(...)

import { tap } from 'rxjs/operators'

obs$.pipe(
  tap((v) => console.log('after step1:', v)),
  map(...),
  tap((v) => console.log('after step2:', v)),
  filter(...),
  tap({
    next: (v) => console.log('next:', v),
    error: (e) => console.log('error:', e),
    complete: () => console.log('done'),
  }),
)

难度
 │
 │  ┌────── ←── 这里很多人放弃
 │  │
 │  │
 │__│
 │
 └─────────────────→ 时间
   入门  半个月    一个月+