RxJS中如何实现带异步清理逻辑的图书借还数据流？

安全阿岳

2026-7-20

解决方法

要处理异步归还操作，确保切换借阅时等待前一本图书归还完成再执行新借阅，可通过自定义Observable结合switchMap实现，核心是在订阅时执行借阅，取消订阅时等待归还操作完成：

import { Observable, Subject } from 'rxjs';
import { switchMap } from 'rxjs/operators';

// 假设已实例化LibraryService
const libraryService = new LibraryService();

const bookId$ = new Subject<string>();

const book$ = bookId$.pipe(
  switchMap(id => 
    new Observable<Book>(async subscriber => {
      // 执行借阅操作
      const book = await libraryService.borrowAsync(id);
      subscriber.next(book);
      
      // 清理函数：取消订阅时等待归还完成
      return async () => {
        await libraryService.returnAsync(id);
      };
    })
  )
);

关键说明

用new Observable封装异步逻辑，让清理函数支持await，确保归还操作完成后再处理下一个借阅请求。
switchMap会在前一个Observable取消订阅时，等待其清理函数的异步操作结束，再订阅新的Observable，满足"同一时间仅借一本"和"所有借阅必须归还"的规则。

也可以用defer结合finalize实现更简洁的写法：

import { defer, Subject } from 'rxjs';
import { switchMap, finalize } from 'rxjs/operators';

const book$ = bookId$.pipe(
  switchMap(id => 
    defer(async () => libraryService.borrowAsync(id)).pipe(
      finalize(async () => {
        await libraryService.returnAsync(id);
      })
    )
  )
);