JavaScript数据流革命：响应式编程与RxJS实战指南

一、响应式编程核心概念

import { fromEvent, interval, combineLatest } from 'rxjs';
import { map, filter, throttleTime } from 'rxjs/operators';

// 创建DOM事件Observable
const button = document.getElementById('searchBtn');
const input = document.getElementById('searchInput');

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

// 组合操作
const search$ = combineLatest([
  input$.pipe(
    map(e => e.target.value),
    filter(text => text.length > 2),
    throttleTime(500)
  ),
  click$
]).pipe(
  map(([query]) => query)
);

// 订阅数据流
search$.subscribe(query => {
  console.log('搜索请求:', query);
  fetchResults(query);
});

// 自动完成示例
input$.pipe(
  map(e => e.target.value),
  filter(text => text.length > 1),
  debounceTime(300),
  distinctUntilChanged(),
  switchMap(query => fetchSuggestions(query))
).subscribe(suggestions => {
  showSuggestions(suggestions);
});

核心优势：数据流管理、声明式编程、自动取消、组合能力

二、高级数据流模式

1. 状态管理集成

import { BehaviorSubject, scan } from 'rxjs';

// 全局状态存储
const initialState = {
  count: 0,
  todos: []
};

const store = new BehaviorSubject(initialState);
const state$ = store.asObservable();

// Reducer函数
function reducer(state, action) {
  switch (action.type) {
    case 'INCREMENT':
      return { ...state, count: state.count + 1 };
    case 'ADD_TODO':
      return { 
        ...state, 
        todos: [...state.todos, action.payload]
      };
    default:
      return state;
  }
}

// 创建action流
const action$ = new Subject();

// 状态更新管道
action$.pipe(
  scan((state, action) => reducer(state, action), initialState)
).subscribe(store);

// 组件中使用
state$.subscribe(state => {
  renderCounter(state.count);
  renderTodos(state.todos);
});

// 触发action
function increment() {
  action$.next({ type: 'INCREMENT' });
}

2. WebSocket实时数据

import { webSocket } from 'rxjs/webSocket';

// 创建WebSocket连接
const socket$ = webSocket({
  url: 'wss://api.example.com/realtime',
  protocol: 'json'
});

// 消息处理
const messages$ = socket$.pipe(
  retryWhen(errors => 
    errors.pipe(
      tap(err => console.log('错误:', err)),
      delayWhen(() => timer(1000))
    )
  ),
  share()
);

// 订阅不同消息类型
const notifications$ = messages$.pipe(
  filter(msg => msg.type === 'notification')
);

const updates$ = messages$.pipe(
  filter(msg => msg.type === 'data_update')
);

// 发送消息
function sendMessage(content) {
  socket$.next({
    type: 'user_message',
    content,
    timestamp: Date.now()
  });
}

// 自动重连机制
socket$.subscribe({
  error: err => console.error('连接中断:', err),
  complete: () => console.log('连接关闭')
});

三、性能优化策略

1. 高效事件处理

// 滚动优化
fromEvent(window, 'scroll').pipe(
  throttleTime(100),
  auditTime(50),
  distinctUntilChanged()
).subscribe(updateUI);

// 拖拽优化
const drag$ = fromEvent(dragElement, 'mousedown').pipe(
  switchMap(start => {
    return fromEvent(document, 'mousemove').pipe(
      map(move => ({
        x: move.clientX - start.clientX,
        y: move.clientY - start.clientY
      })),
      takeUntil(fromEvent(document, 'mouseup'))
  })
);

// 多事件合并
const form$ = combineLatest([
  fromEvent(nameInput, 'input').pipe(map(e => e.target.value)),
  fromEvent(emailInput, 'input').pipe(map(e => e.target.value)),
  fromEvent(termsCheckbox, 'change').pipe(map(e => e.target.checked))
]).pipe(
  debounceTime(300)
);

2. 内存泄漏防护

// 自动取消订阅
class Component {
  constructor() {
    this.subscriptions = new Subscription();
    
    const timer$ = interval(1000).pipe(
      takeUntil(this.destroy$)
    );
    
    this.subscriptions.add(
      timer$.subscribe(console.log)
    );
  }
  
  destroy() {
    this.subscriptions.unsubscribe();
  }
}

// 高阶Observable管理
function search(query) {
  return ajax.getJSON(`/api/search?q=${query}`).pipe(
    timeout(5000),
    catchError(err => of({ error: err.message }))
  );
}

input$.pipe(
  debounceTime(300),
  distinctUntilChanged(),
  switchMap(query => 
    query ? search(query) : EMPTY
  )
).subscribe(showResults);

四、电商实时看板实战

1. 实时数据可视化

// 合并多个数据源
const dashboard$ = merge(
  orderUpdates$.pipe(map(data => ({ type: 'orders', data }))),
  userActivity$.pipe(map(data => ({ type: 'activity', data }))),
  inventoryUpdates$.pipe(map(data => ({ type: 'inventory', data })))
).pipe(
  scan((acc, curr) => {
    acc[curr.type] = curr.data;
    return acc;
  }, {})
);

// 实时渲染
dashboard$.subscribe(data => {
  renderOrderChart(data.orders);
  updateActivityFeed(data.activity);
  refreshInventory(data.inventory);
});

// 错误处理专用流
const errors$ = merge(
  orderUpdates$.pipe(
    catchError(err => of({ type: 'order_error', err }))
  ),
  userActivity$.pipe(
    catchError(err => of({ type: 'activity_error', err }))
  )
).pipe(
  filter(msg => msg.err)
);

errors$.subscribe(showErrorAlert);