如何在NestJS中从微服务返回的Observable提取值？

你遇到的核心问题是NestJS微服务ClientProxy.send返回的是Observable异步流，同步调用无法直接获取结果，必须通过异步方式（结合RxJS的lastValueFrom和async/await）处理。以下是针对你代码的具体修复方案：

1. 修复调用方（消费者）的控制器代码 #

原控制器存在未正确获取请求头、方法未异步处理、代码结构错误等问题，修改后：

// app.controller.ts
import { Body, Controller, Get, Post, Headers, UnauthorizedException } from '@nestjs/common';
import { AppService } from './app.service';
import { PayloadDTO} from './payload.dto';

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Get()
  getHello(): string {
    return this.appService.getHello();
  }

  @Post()
  async createUserData(@Body() payload: PayloadDTO, @Headers() headers) {
    // 校验并提取Bearer Token
    const authHeader = headers.authorization;
    if (!authHeader || !authHeader.startsWith('Bearer ')) {
      throw new UnauthorizedException('缺少有效Token');
    }
    const token = authHeader.replace('Bearer ', '');
    
    // 异步获取Token验证结果
    const isValidToken = await this.appService.validateToken(token);
    
    if(isValidToken) {
      await this.appService.createRecord(payload); // 若createRecord为异步操作需加await
      return { message: '记录创建成功' };
    } else {
      throw new UnauthorizedException('Token无效');
    }
  }
}

2. 修复调用方的服务代码 #

修改validateToken方法，使用lastValueFrom将Observable转为Promise，异步返回验证结果：

// app.service.ts
import { Inject, Injectable } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
import { PayloadDTO } from './payload.dto';
import { lastValueFrom } from 'rxjs'; // 导入RxJS工具方法

@Injectable()
export class AppService {

  constructor(
    @Inject('COMMUNICATION') private readonly commClient: ClientProxy
  ) {}

 // 异步验证Token，返回Promise<boolean>
  async validateToken(token: string): Promise<boolean> {
    // 调用微服务并将Observable转为Promise
    const checkToken$ = this.commClient.send({cmd:'validate_token'}, token);
    const isValid = await lastValueFrom(checkToken$);
    return isValid; 
  } 

  // 若创建记录涉及异步操作，建议改为async方法
  async createRecord(payload: PayloadDTO) {
    // 记录创建逻辑
  }
}

3. 修复验证Token的微服务控制器 #

原控制器未接收调用方传递的Token参数，导致无法完成验证逻辑，修改后：

// 微服务的app.controller.ts
import { Controller } from '@nestjs/common';
import { MessagePattern } from '@nestjs/microservices';
import { AppService } from './app.service';

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  // 接收Token参数并传递给业务方法
  @MessagePattern({ cmd: 'validate_token' })
  validateToken(token: string) {
    return this.appService.checkToken(token);
  }
}

关键说明 #

• lastValueFrom会等待Observable流完成并获取最后一个值，适配微服务单次响应的场景
• 所有涉及异步操作的方法必须标记为async，并用await获取结果
• 微服务的MessagePattern方法必须接收调用方传递的参数，否则无法拿到待验证的Token

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