如何在Express.js常规中间件中消费RxJS Observable？

核心思路是利用RxJS的订阅机制适配Express的回调式中间件模型——因为Observable是惰性数据流，必须订阅才会触发执行，在数据流的next回调中处理业务逻辑并调用Express的next()，同时必须处理错误避免请求挂起。

基础实现示例 #

修改你的中间件代码，通过订阅getSettings(req)返回的Observable来处理：

middleware(req, res, next) {
  getSettings(req).subscribe({
    next: (settings) => {
      // 处理settings，比如挂载到req对象供后续中间件/路由使用
      req.settings = settings;
      // 执行后续流程
      next();
    },
    error: (err) => {
      // 将错误传递给Express的错误处理中间件
      next(err);
    }
  });
}

处理多个Observable场景 #

如果需要同时获取多个Observable数据（比如配置+用户信息），可以用forkJoin或combineLatest合并后订阅：

import { forkJoin } from 'rxjs';

middleware(req, res, next) {
  const settings$ = getSettings(req);
  const userInfo$ = getUserInfo(req);

  forkJoin([settings$, userInfo$]).subscribe({
    next: ([settings, userInfo]) => {
      req.settings = settings;
      req.userInfo = userInfo;
      next();
    },
    error: (err) => next(err)
  });
}

关键注意事项 #

• 必须订阅Observable：Observable是惰性的，不订阅的话getSettings不会执行任何逻辑
• 务必处理错误回调：如果Observable抛出错误但未处理，请求会一直处于挂起状态，无法返回响应
• 避免内存泄漏：如果getSettings是会持续发射值的热Observable，建议用first()或take(1)操作符确保只取一次值就完成订阅：

  import { first } from 'rxjs/operators';
  
  middleware(req, res, next) {
    getSettings(req).pipe(first()).subscribe({
      next: (settings) => {
        req.settings = settings;
        next();
      },
      error: (err) => next(err)
    });
  }
  

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