RxJS中Observable对应Redux getState的等效实现方法

嘿，很高兴看到你在啃RxJS和响应式编程的知识点！针对你问的「RxJS里对应Redux getState的等效实现」，其实核心是要在RxJS的流中保存并能随时获取当前状态的快照，下面给你分享两个最常用的方案：

1. 用BehaviorSubject作为状态容器 #

Redux的getState本质是获取当前状态的快照，而RxJS里的BehaviorSubject天生就具备这个能力——它会始终保存最新的发射值，新订阅者会立刻收到当前的最新值，同时你可以直接通过它的.value属性同步获取当前状态，这完全对应Redux的getState。

举个贴合你示例的用法：

import Rx from 'rxjs'

// 初始化状态（类似Redux的initialState）
const initialState = {
  data: [],
  loading: false,
  error: null
}

// 创建BehaviorSubject作为状态容器
const stateSubject = new Rx.BehaviorSubject(initialState)

// 封装一个类似getState的函数
const getState = () => stateSubject.value

// 模拟你的fetchData逻辑，更新状态
const fetchData = (first, last) => {
  // 先更新为加载中状态
  stateSubject.next({...getState(), loading: true })
  
  return Rx.Observable.range(first, last)
    .flatMap(async n => {
      // 模拟异步请求
      const res = await fetch(`https://example.com/data/${n}`)
      return res.json()
    })
    .toArray()
    .subscribe({
      next: (data) => {
        // 请求成功，更新状态
        stateSubject.next({...getState(), loading: false, data })
      },
      error: (err) => {
        // 请求失败，更新状态
        stateSubject.next({...getState(), loading: false, error: err.message })
      }
    })
}

// 订阅状态变化（类似Redux的subscribe）
stateSubject.subscribe(state => {
  console.log('当前状态:', state)
  // 这里可以更新DOM，比如控制loader显示
  if (state.loading) {
    loader.style.display = 'block'
  } else {
    loader.style.display = 'none'
    domContainer.innerHTML = JSON.stringify(state.data)
  }
})

// 点击按钮触发请求
Rx.Observable.fromEvent(clickElement, 'click')
  .subscribe(() => fetchData(1, 5))

2. 结合scan操作符与BehaviorSubject #

如果你习惯Redux里用reducer处理状态更新的模式，可以用scan操作符来累积状态（完全对应Redux的reducer逻辑），然后把scan的输出订阅到BehaviorSubject上，这样既可以通过流来处理状态变更，又能随时获取当前状态快照：

import Rx from 'rxjs'

// 定义reducer（和Redux的reducer完全一致）
const reducer = (state, action) => {
  switch (action.type) {
    case 'FETCH_START':
      return {...state, loading: true }
    case 'FETCH_SUCCESS':
      return {...state, loading: false, data: action.payload }
    case 'FETCH_ERROR':
      return {...state, loading: false, error: action.payload }
    default:
      return state
  }
}

// 初始化状态
const initialState = { data: [], loading: false, error: null }

// 创建action流（类似Redux的dispatch）
const action$ = new Rx.Subject()

// 用scan生成状态流
const state$ = action$.scan(reducer, initialState)

// 把状态流订阅到BehaviorSubject，方便获取当前状态
const stateSubject = new Rx.BehaviorSubject(initialState)
state$.subscribe(stateSubject)

// 封装getState
const getState = () => stateSubject.value

// 封装dispatch
const dispatch = (action) => action$.next(action)

// 模拟fetchData逻辑
const fetchData = (first, last) => {
  dispatch({ type: 'FETCH_START' })
  
  Rx.Observable.range(first, last)
    .flatMap(async n => {
      const res = await fetch(`https://example.com/data/${n}`)
      return res.json()
    })
    .toArray()
    .subscribe({
      next: (data) => dispatch({ type: 'FETCH_SUCCESS', payload: data }),
      error: (err) => dispatch({ type: 'FETCH_ERROR', payload: err.message })
    })
}

// 订阅状态更新
state$.subscribe(state => {
  console.log('当前状态:', state)
  // 更新DOM逻辑...
})

// 点击触发
Rx.Observable.fromEvent(clickElement, 'click')
  .subscribe(() => fetchData(1, 5))

关键说明 #

• 不要用普通的Observable来做状态容器，因为它是「冷流」，没有状态保存能力，每次订阅都会从头执行，无法获取当前快照。
• BehaviorSubject的.value属性是同步获取的，这和Redux的getState完全一致；如果你需要异步获取最新状态，可以订阅它的流。

内容的提问来源于stack exchange，提问作者Lesbaa