WHAT - RxJS 异步事件流处理

RxJS 是一个用于处理异步事件流的库，常用于 响应式编程。它提供了 Observables（可观察对象），以及一系列 操作符（Operators） 来处理数据流。

适用于 事件处理、网络请求、WebSocket、定时操作等。

RxJS 关键概念

1. Observable（可观察对象）

   • 可以看作是一个数据流，比如点击事件、HTTP 请求等。
   • 通过 .subscribe() 订阅，获取数据流的值。

2. Observer（观察者）

   • 订阅 Observable 的对象，包含 next、error 和 complete 方法。

3. Subscription（订阅）

   • 订阅 Observable 后，返回 Subscription，可用于取消订阅。

4. Operators（操作符）

   • 用于对数据流进行转换、过滤、合并等操作。
   • 主要分为 创建、转换、组合、过滤 等类型的操作符。

5. Subject（主题）

   • 既是 Observable 也是 Observer，可以手动推送数据。

简单示例

import { Observable } from 'rxjs';

// 创建 Observable
const observable = new Observable(subscriber => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  subscriber.complete();
});

// 订阅 Observable
observable.subscribe({
  next: value => console.log(value),
  complete: () => console.log('完成'),
});

// 输出：
// 1
// 2
// 3
// 完成

常用操作符

1. 创建 Observables

import { of, from } from 'rxjs';

of(1, 2, 3).subscribe(console.log);  // 依次输出 1, 2, 3
from([4, 5, 6]).subscribe(console.log);  // 依次输出 4, 5, 6

2. 转换操作符（map）

import { of } from 'rxjs';
import { map } from 'rxjs/operators';

of(1, 2, 3)
  .pipe(map(x => x * 2))
  .subscribe(console.log);  // 输出 2, 4, 6

3. 过滤操作符（filter）

import { of } from 'rxjs';
import { filter } from 'rxjs/operators';

of(1, 2, 3, 4, 5)
  .pipe(filter(x => x % 2 === 0))
  .subscribe(console.log);  // 输出 2, 4

4. 合并多个流（mergeMap）

import { of } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

of('A', 'B')
  .pipe(mergeMap(val => of(val + '1', val + '2')))
  .subscribe(console.log);  
// 输出：A1 A2 B1 B2

适用于 React & TypeScript

在 React 里，RxJS 主要用于：

• 管理状态流（类似 Redux，但是流式）
• 处理 WebSocket 或轮询
• 流式处理用户输入（如搜索框防抖）

React 组件中使用 RxJS

import { useEffect, useState } from 'react';
import { fromEvent } from 'rxjs';
import { map, debounceTime } from 'rxjs/operators';

export default function RxJSExample() {
  const [text, setText] = useState('');

  useEffect(() => {
    const input = document.getElementById('myInput') as HTMLInputElement;
    const subscription = fromEvent(input, 'input')
      .pipe(
        debounceTime(500), // 防抖
        map(event => (event.target as HTMLInputElement).value)
      )
      .subscribe(value => setText(value));

    return () => subscription.unsubscribe();
  }, []);

  return (
    <div>
      <input id="myInput" type="text" placeholder="输入内容..." />
      <p>输入: {text}</p>
    </div>
  );
}