使用 RxJS 实现可靠的异步搜索框

本文源于张乐聪同学的分享，在原有代码基础上进行了一些修改

前言

异步搜索框是一个业务中非常常见的诉求，但是想实现一个可靠的异步搜索框却不是一个简单的任务，为了使其可靠（性能好 + Bug 少 + 体验好 + 易维护），实现者需要考虑非常多的方面.

异步搜索框的难点

1. 针对于搜索做 debounce 操作，在用户的输入过程中不立即搜索（性能好，节省网络资源）
2. 对于输入为空的时候不进行 debounce（体验好，从有搜索内容到无搜索内容立即响应）
3. 对于 debounce 后的输入去重，不发送重复请求，例如从 a -> ab（debounce 掉，不发送）-> a，可能对 a 发送两次搜索请求（性能好，节省网络资源）
4. 正确处理时序，不要被早发送的请求响应覆盖晚发送的请求响应（体验好）
5. 正确处理异常（体验好）
6. 正确处理 loading，只要还有请求没有返回就维持 loading（体验好）
7. 在正确实现之前所有需求的前提下维持实现的可维护性（易维护 + 不容易出 Bug）

常见实现的问题

React JSX
Copy
function SearchBox() {
  const [result, setResult] = useState("")
  const handleInput = (e) => {
    const value = e.target.value
    request(value).then((response) => {
      setResult(response.data)
    })
  }
  return (
    <>
      <input onChange={handleInput} />
      {result}
    </>
  )
}

这种实现最典型的问题是时序问题不能被正确的处理，没有个先来后到的讲究，谁来谁覆盖。 因此要进行处理的话要么维持发送时间，要么记下来发送的内容，来确保响应可以和请求匹配。

React JSX
Copy
// 产生 4000 2000 1000 4000 ...
const random2 = (function () {
  let i = 1

  const array = [4000, 2000, 1000]

  return () => {
    return array[i++ % array.length]
  }
})()

function request(value) {
  return new Promise<{
    data: string
  }>((resolve) => {
    setTimeout(() => {
      resolve({
        data: value,
      })
    }, random2())
  })
}

function SearchBox() {
  const [result, setResult] = useState("")
  const latestRequestTimeRef = useRef(0)
  const handleInput = (e) => {
    const value = e.target.value
    const requestTime = Date.now() // 记录时间
    latestRequestTimeRef.current = requestTime
    request(value).then((response) => {
      if (requestTime >= latestRequestTimeRef.current) {
        // 对比时间
        setResult(response.data)
      }
    })
  }
  return (
    <>
      <input onChange={handleInput} />
      {result}
    </>
  )
}

如果涉及 debounce，通常我们都会直接使用工具函数比如 lodash 的 debounce，它无法实现条件 debounce，因此我们需要自己专门实现。

即使过了这关，在后续的 error、loading 处理中，你会发现，所有的代码都挤在 handleInput 中，状态相互纠缠。不光可靠性难以保证、持续维护的难度也会越来越大。

可靠实现的难度在哪？

如果你有一些编写异步操作的经验，会发现每增加一个 feature 都需要维护一些状态、并且由于逻辑关联，会和原有的逻辑搅在一起，就像一个线团一样。在没有高层次抽象的情况下，很难将不同的异步 feature 进行隔离。随着功能的增多，这个线团越来越大、越来越乱，直到艰难维护、崩溃、重写或者消亡。

所以解决问题的一个思路就是：将不同的 feature 以解耦、内聚的形式实现，相互独立，各自维护，再统一串联。

1. 引入 RxJS 做一个输入的流

通过制造一个流，在输入值改变的时候向流发送数据，并监听这个流，可以将输入内容实时的同步在页面上。

React JSX
Copy
function SearchBox() {
  const [result, setResult] = useState("")
  // 下面的 BehaviorSubject 和 Subject 一模一样，除了有一个初始值会在订阅时立刻发出
  const input$ = useMemo(() => new BehaviorSubject(""), [])
  // 输入内容时向流发送值
  const handleInput = (e) => {
    input$.next(e.target.value)
  }
  useEffect(() => {
    // 订阅这个流
    const subscription = input$.subscribe((v) => {
      setResult(v)
    })
    return () => {
      // 组件卸载时取消订阅
      subscription.unsubscribe()
    }
  }, [])

  return (
    <>
      <input onChange={handleInput} />
      {result}
    </>
  )
}

2. debounce

第一步我们先进行 debounce 的实现，在搜索值为空的时候立即响应，其他情况下 debounce： 我们利用 debounce 操作符，在输入值为空字符串的时候立马发送值，在输入不为空的时候等待 500ms 再发送值。

React JSX
Copy
function SearchBox() {
  const [result, setResult] = useState("")
  // 下面的 BehaviorSubject 和 Subject 一模一样，除了有一个初始值会在订阅时立刻发出
  const input$ = useMemo(() => new BehaviorSubject(""), [])
  // 输入内容时向流发送值
  const handleInput = (e) => {
    input$.next(e.target.value)
  }
  useEffect(() => {
    const subscription = input$
      .pipe(
        // 防抖的实现 -----------------------------
        debounce((input) => {
          if (input.length === 0) {
            return of(null) // 立即响应
          } else {
            return timer(500) // 等待 500ms
          }
        })
      )
      .subscribe((v) => {
        setResult(v)
      })
    return () => {
      subscription.unsubscribe()
    }
  }, [])

  return (
    <>
      <input onChange={handleInput} />
      {result}
    </>
  )
}

3. 去重

React JSX
Copy
function SearchBox() {
  const [result, setResult] = useState("")
  // 下面的 BehaviorSubject 和 Subject 一模一样，除了有一个初始值会在订阅时立刻发出
  const input$ = useMemo(() => new BehaviorSubject(""), [])
  // 输入内容时向流发送值
  const handleInput = (e) => {
    input$.next(e.target.value)
  }
  useEffect(() => {
    const subscription = input$
      .pipe(
        // 防抖的实现 -----------------------------
        debounce((input) => {
          if (input.length === 0) {
            return of(null) // 立即响应
          } else {
            return timer(500) // 等待 500ms
          }
        }),
        distinctUntilChanged()
      )
      .subscribe((v) => {
        setResult(v)
      })
    return () => {
      subscription.unsubscribe()
    }
  }, [])

  return (
    <>
      <input onChange={handleInput} />
      {result}
    </>
  )
}

4. 网络请求 + 时序处理

Rxjs 提供了 switchMap 操作符来完成 Promise 到值的解包过程和异步时序控制能力。switchMap 可以将一个流映射为新的流，我们可以将一个文本流通过 Promise 映射为一个文本流到 Promise resolve 结果的流，同时 switchMap 还有一个特殊的能力就是会丢弃掉比最新输入发起时间晚到的值：

React JSX
Copy
// 产生 4000 2000 1000 4000 ...
const random2 = (function () {
  let i = 1

  const array = [4000, 2000, 1000]

  return () => {
    return array[i++ % array.length]
  }
})()

function request(value) {
  return new Promise<{
    data: string
  }>((resolve) => {
    setTimeout(() => {
      resolve({
        data: value,
      })
    }, random2())
  })
}

function SearchBox() {
  const [result, setResult] = useState("")
  // 下面的 BehaviorSubject 和 Subject 一模一样，除了有一个初始值会在订阅时立刻发出
  const input$ = useMemo(() => new BehaviorSubject(""), [])
  // 输入内容时向流发送值
  const handleInput = (e) => {
    input$.next(e.target.value)
  }
  useEffect(() => {
    const subscription = input$
      .pipe(
        // 防抖的实现 -----------------------------
        debounce((input) => {
          if (input.length === 0) {
            return of(null) // 立即响应
          } else {
            return timer(500) // 等待 500ms
          }
        }),
        distinctUntilChanged(),
        // 网络请求的实现 -----------------------------
        switchMap((input) => {
          return request(input) // 取最新开发发起的结果
        })
      )
      .subscribe((v) => {
        setResult(v.data)
      })
    return () => {
      subscription.unsubscribe()
    }
  }, [])

  return (
    <>
      <input onChange={handleInput} />
      {result}
    </>
  )
}

5. Loading + 异常处理

React JSX
Copy
function fetcher(input: string): Promise<{
  value: string
  error: boolean
}> {
  return new Promise<string>((resolve, reject) => {
    setTimeout(() => {
      if (Math.random() > 0.5) reject()
      resolve("api " + input)
    }, Math.random() * 1000)
  })
    .then((res) => {
      return {
        value: res,
        error: false,
      }
    })
    .catch(() => {
      return {
        value: "",
        error: true,
      }
    })
}

function SearchBox() {
  const [result, setResult] = useState("")
  // 下面的 BehaviorSubject 和 Subject 一模一样，除了有一个初始值会在订阅时立刻发出
  const input$ = useMemo(() => new BehaviorSubject(""), [])
  const [loading, setLoading] = useState(false)
  const [error, setError] = useState(false)

  const errorRef = useRef<boolean>(false)
  errorRef.current = error

  // 输入内容时向流发送值
  const handleInput = (e) => {
    input$.next(e.target.value)
  }
  useEffect(() => {
    const subscription = input$
      .pipe(
        // 防抖的实现 -----------------------------
        debounce((input) => {
          // 补充 error 处理
          if (input.length === 0 || errorRef.current) {
            return of(null) // 立即响应
          } else {
            return timer(500) // 等待 500ms
          }
        }),
        distinctUntilChanged(),
        // 网络请求的实现 -----------------------------
        switchMap((input) => {
          if (input.length === 0) {
            setLoading(false)
            setError(false)
            return of({
              value: "default",
              error: false,
            })
          }
          setError(false)
          setLoading(true)
          return fetcher(input)
        })
      )
      .subscribe({
        next: ({ error, value }) => {
          if (error) {
            setError(true)
            setLoading(false)
          } else {
            setError(false)
            setLoading(false)
            setResult(value)
          }
        },
      })
    return () => {
      subscription.unsubscribe()
    }
  }, [])

  return (
    <>
      <div>
        <input onChange={handleInput} />
      </div>
      {error ? "error" : loading ? "loading" : result}
    </>
  )
}

最后

代码可见 https://github.com/sedationh/search-box