NestJS/NATS微服务异常：内部发射事件无法被跨服务接收

核心原因 #

微服务A通过普通NATS客户端发送事件，而非JetStream客户端，导致事件未进入JetStream流，而微服务B是通过JetStream订阅流中的事件，因此无法接收普通pub/sub消息。

具体问题点及修复步骤 #

1. 微服务A使用了普通ClientProxy而非NatsJetStreamClientProxy
   JetStream是NATS的持久化消息系统，与核心pub/sub是分离的。普通emit发送的消息不会存入JetStream流，而微服务B通过@EventPattern结合JetStream配置订阅的是流内消息，自然无法接收。

   修复:
   在微服务A的模块中注册JetStream客户端，并注入NatsJetStreamClientProxy：

   // microservice-a.module.ts
   import { Module } from '@nestjs/common';
   import { ClientsModule } from '@nestjs/microservices';
   import { NatsJetStreamTransport } from '@nestjs/microservices/external/nats-jetstream.transport';
   import { MicroserviceAController } from './microservice-a.controller';
   import { BusinessRulesService } from './business-rules.service';
   
   @Module({
     imports: [
       ClientsModule.register([
         {
           name: 'JET_STREAM_CLIENT',
           transport: NatsJetStreamTransport,
           options: {
             servers: 'nats://localhost:4222', // 匹配你的NATS服务器地址
             stream: {
               name: 'business-rules-stream',
               subjects: ['businessRule.apply', 'businessRule.appliedResult'], // 包含目标事件主题
             },
           },
         },
       ]),
     ],
     controllers: [MicroserviceAController],
     providers: [BusinessRulesService],
   })
   export class MicroserviceAModule {}
   

2. 修正微服务A控制器中的客户端注入与调用
   将普通ClientProxy替换为NatsJetStreamClientProxy，并确保发送事件时使用该客户端：

   // microservice-a.controller.ts
   import { Controller, Inject } from '@nestjs/common';
   import { EventPattern, Payload, Ctx } from '@nestjs/microservices';
   import { NatsJetStreamContext, NatsJetStreamClientProxy } from '@nestjs/microservices';
   import { BusinessRulesService } from './business-rules.service';
   
   @Controller()
   export class MicroserviceAController {
     constructor(
       private readonly businessRulesService: BusinessRulesService,
       @Inject('JET_STREAM_CLIENT') private readonly jetStreamClient: NatsJetStreamClientProxy,
     ) {}
   
     @EventPattern("businessRule.apply")
     async applyRule(
       @Payload() { ruleName, input }: { ruleName: string; input: any },
       @Ctx() context: NatsJetStreamContext,
     ): Promise<any> {
       const result = await this.businessRulesService.applyRule(ruleName, input);
       // 使用JetStream客户端发送事件
       await this.jetStreamClient.emit("businessRule.appliedResult", result);
       context.message.ack();
       return;
     }
   }
   

3. 额外检查项

   • 确认所有服务（REST API、微服务A、微服务B）连接到同一NATS服务器
   • 微服务B的JetStream配置中，流必须包含businessRule.appliedResult主题
   • 查看NATS服务器日志，排查流创建或消息发送时的错误

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