NestJS 从入门到精通 05：微服务架构拆解 - gRPC 与 Kafka 分布式系统实战

当单体应用遇到 高并发瓶颈、服务雪崩风险 或 多语言协作难题 时，微服务架构成为必然选择。在本期，我们将深入探索 NestJS 微服务生态系统，构建可横向扩展的分布式系统。

一、NestJS 微服务架构核心概念

1. 微服务通信模式对比

2. NestJS 微服务体系架构

graph TD
    A[API Gateway] -->|gRPC| B[用户服务]
    A -->|gRPC| C[订单服务]
    A -->|gRPC| D[支付服务]
    C -->|Kafka| E[库存服务]
    C -->|Kafka| F[通知服务]
    D -->|Kafka| G[风控服务]

二、gRPC 实战：跨服务高效通信

1. 安装核心依赖

npm install @nestjs/microservices @grpc/proto-loader @grpc/grpc-js

2. 定义 Protobuf 协议

// user.proto
syntax = "proto3";

package user;

service UserService {
  rpc GetUserInfo (UserRequest) returns (UserReply) {}
}

message UserRequest {
  int32 id = 1;
}

message UserReply {
  int32 id = 1;
  string name = 2;
  string email = 3;
}

3. 实现服务端

// main.ts - 用户服务
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { join } from 'path';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
    transport: Transport.GRPC,
    options: {
      url: '0.0.0.0:50051',
      package: 'user',
      protoPath: join(__dirname, 'user.proto'),
    },
  });
  await app.listen();
}

4. 客户端调用

// order.service.ts - 订单服务
import { Inject, Injectable, OnModuleInit } from '@nestjs/common';
import { ClientGrpc } from '@nestjs/microservices';

@Injectable()
export class OrderService implements OnModuleInit {
  private userService: UserService;
  
  constructor(@Inject('USER_PACKAGE') private client: ClientGrpc) {}
  
  onModuleInit() {
    this.userService = this.client.getService<UserService>('UserService');
  }
  
  async getOrderDetail(orderId: number) {
    const userInfo = await firstValueFrom(
      this.userService.GetUserInfo({ id: 1 })
    );
    return { orderId, userInfo };
  }
}

// order.module.ts
@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'USER_PACKAGE',
        transport: Transport.GRPC,
        options: {
          package: 'user',
          protoPath: join(__dirname, 'user.proto'),
        },
      },
    ]),
  ],
})
export class OrderModule {}

三、Kafka 实战：构建分布式事件总线

1. Kafka 架构核心概念

graph LR
    A[生产者] -->|发布消息| B[Kafka Broker]
    B --> C[Topic]
    C --> D[分区1]
    C --> E[分区2]
    D -->|订阅| F[消费者组1]
    E -->|订阅| G[消费者组2]

2. 配置 Kafka 微服务

// kafka.service.ts
import { Kafka, Producer, Consumer, ProducerRecord } from 'kafkajs';

@Injectable()
export class KafkaService {
  private kafka: Kafka;
  private producer: Producer;
  private consumers = new Map<string, Consumer>();
  
  constructor() {
    this.kafka = new Kafka({
      clientId: 'order-service',
      brokers: ['kafka1:9092', 'kafka2:9092'],
    });
    this.producer = this.kafka.producer();
  }
  
  async connect() {
    await this.producer.connect();
  }
  
  async send(topic: string, payload: any) {
    await this.producer.send({
      topic,
      messages: [{ value: JSON.stringify(payload) }],
    });
  }
  
  async subscribe(topic: string, groupId: string, callback: (msg) => void) {
    const consumer = this.kafka.consumer({ groupId });
    await consumer.connect();
    await consumer.subscribe({ topic, fromBeginning: true });
    
    await consumer.run({
      eachMessage: async ({ message }) => {
        callback(JSON.parse(message.value.toString()));
      },
    });
    
    this.consumers.set(`${topic}-${groupId}`, consumer);
  }
}

3. Kafka 消息处理示范

订单服务（生产者）:

// order.service.ts
@Injectable()
export class OrderService {
  constructor(private kafka: KafkaService) {}
  
  async createOrder(order: OrderDto) {
    // 保存订单到数据库...
    await this.kafka.send('order-created', order);
  }
}

库存服务（消费者）:

// inventory.service.ts
@Injectable()
export class InventoryService {
  constructor(private kafka: KafkaService) {}
  
  onModuleInit() {
    this.kafka.subscribe('order-created', 'inventory-group', async (order) => {
      await this.deductStock(order.items);
    });
  }
  
  private async deductStock(items: Item[]) {
    // 库存扣减实现...
  }
}

四、分布式事务挑战：Saga 模式实战

1. 订单创建场景的事务需求

sequenceDiagram
  participant 订单服务
  participant 库存服务
  participant 支付服务
  订单服务->>库存服务: 扣减库存
  库存服务-->>订单服务: 库存已扣
  订单服务->>支付服务: 扣款请求
  支付服务-->>订单服务: 扣款成功
  订单服务->>用户服务: 添加积分

2. Saga 补偿事务实现

// order.saga.ts
enum OrderSagaState {
  START,
  RESERVE_STOCK,
  PROCESS_PAYMENT,
  ADD_POINTS,
  COMPLETED,
}

@Injectable()
export class OrderSaga {
  private stateMachine: StateMachine<OrderSagaState, any>;

  constructor(
    private kafka: KafkaService,
    private orderRepo: OrderRepository
  ) {
    this.initSaga();
  }
  
  private initSaga() {
    this.stateMachine = new StateMachine<OrderSagaState, any>({
      [OrderSagaState.START]: {
        on: { 
          ORDER_CREATED: { 
            target: OrderSagaState.RESERVE_STOCK,
            action: this.reserveStock.bind(this) 
          } 
        }
      },
      [OrderSagaState.RESERVE_STOCK]: {
        on: { 
          STOCK_RESERVED: OrderSagaState.PROCESS_PAYMENT,
          STOCK_FAILED: OrderSagaState.CANCEL_ORDER
        }
      },
      // ...其他状态转换
    });
  }
  
  private async reserveStock(orderId: number) {
    try {
      const order = await this.orderRepo.findById(orderId);
      await this.kafka.send('reserve-stock', { orderId, items: order.items });
    } catch (err) {
      // 触发补偿操作
      this.stateMachine.fire('STOCK_FAILED');
    }
  }
  
  // 补偿动作
  private async cancelOrder(orderId: number) {
    await this.kafka.send('release-stock', { orderId });
    await this.orderRepo.cancelOrder(orderId);
  }
}

3. 事务监控控制台

// saga-monitor.gateway.ts - 实时监控
@WebSocketGateway()
export class SagaMonitorGateway {
  @WebSocketServer()
  server: Server;
  
  @OnEvent('saga.*')
  handleSagaEvent(event: string, data: any) {
    this.server.emit('saga-update', {
      event,
      sagaId: data.sagaId,
      state: data.state,
      timestamp: new Date()
    });
  }
}

五、综合实战：金融账户系统架构

1. 系统组件划分

graph LR
    G[API网关] --- A[账户服务]
    G --- B[交易服务]
    G --- C[风控服务]
    G --- D[通知服务]
    B --> E[Kafka交易事件]
    C -->|消费| E
    D -->|消费| E

2. gRPC 服务接口设计

service AccountService {
  rpc OpenAccount (OpenAccountRequest) returns (Account) {}
  rpc GetAccount (GetAccountRequest) returns (Account) {}
}

service TransactionService {
  rpc TransferFunds (TransferRequest) returns (TransferResponse) {}
  rpc GetTransactionHistory (HistoryRequest) returns (stream Transaction) {}
}

3. 分布式事务处理流程

// 跨服务转账实现
async transfer(transferDto: TransferDto) {
  // 1. 检查账户状态
  const fromAccount = await this.accountClient.GetAccount({
    id: transferDto.fromAccountId,
  });
  
  // 2. 发起转账（Saga启动）
  this.saga.start('TRANSFER', {
    from: transferDto.fromAccountId,
    to: transferDto.toAccountId,
    amount: transferDto.amount,
    currency: transferDto.currency,
  });
  
  // 3. 更新账户
  const result = await this.transactionClient.TransferFunds({
    from: transferDto.fromAccountId,
    to: transferDto.toAccountId,
    amount: transferDto.amount,
    currency: fromAccount.currency, // 保证币种一致
  });
  
  return { transactionId: result.id };
}

4. 实时风控监控流

// 流式风控处理
@EventPattern('transaction.created')
async handleTransactionEvent(data: TransactionEvent) {
  // 1. 异常模式检测
  const riskScore = await this.riskModel.detect(data);
  
  // 2. 高风险交易警报
  if (riskScore > 0.8) {
    await this.kafka.send('risk-alert', {
      transactionId: data.id,
      score: riskScore,
      trigger: 'HIGH_RISK',
    });
    
    // 3. 暂停转账操作（与Saga协作）
    this.saga.pause(data.transactionId);
  }
}

六、性能优化：微服务调优策略

1. gRPC 调优参数

// 服务端配置优化
  options: {
    keepalive: {
      keepaliveTimeMs: 120000,
      keepaliveTimeoutMs: 20000,
      keepalivePermitWithoutCalls: 1,
    },
    maxSendMessageLength: 1024 * 1024 * 100, // 100MB
    maxReceiveMessageLength: 1024 * 1024 * 100,
  }

2. Kafka 消费者级并行处理

// 扩展分区消费能力
async setupConsumers(topic: string, count: number) {
  for (let i = 0; i < count; i++) {
    const consumer = this.kafka.consumer({ groupId });
    await consumer.connect();
    await consumer.subscribe({ topic, fromBeginning: true });
    await consumer.run({ /*处理器逻辑*/ });
    this.consumers.push(consumer);
  }
}

3. 分布式追踪集成（Jaeger）

// 启用OpenTelemetry追踪
import { KafkaInstrumentation } from '@opentelemetry/instrumentation-kafkajs';
import { GrpcInstrumentation } from '@opentelemetry/instrumentation-grpc';
import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';

const provider = new NodeTracerProvider();
provider.register();

provider.addSpanProcessor(new JaegerExporter({
  endpoint: 'http://jaeger:14268/api/traces',
}));

// 注册gRPC和Kafka追踪
registerInstrumentations({
  instrumentations: [
    new GrpcInstrumentation(),
    new KafkaInstrumentation(),
  ],
});

七、避坑指南：分布式系统常见问题

❌ 问题1：gRPC 版本兼容性问题

错误表现：

ERROR @grpc/proto-loader unsupported field type 

解决方案：

1. 确认所有服务使用相同的 Protobuf 版本

2. 添加一致的 gRPC 依赖版本锁定：

   // package.json
   "resolutions": {
     "@grpc/grpc-js": "1.5.5",
     "@grpc/proto-loader": "0.6.9"
   }

❌ 问题2：Kafka 消息丢失

解决方案：

// 启用生产者确认
await producer.send({
  topic: 'orders',
  messages: [{ value: 'order_data' }],
  acks: -1, // 要求所有副本确认
})

❌ 问题3：分布式死锁

模式：服务A等待服务B，服务B等待服务A
解法：实现超时回滚机制

// 设置Saga操作超时
this.stateMachine.withConfig({
  actions: {
    reserveStock: {
      timeout: 5000, // 5秒超时
      onTimeout: 'handleReserveTimeout'
    }
  }
})

八、下期预告：GraphQL API 开发

《GraphQL现代API开发：解决RESTful之痛》 将深入：
➠ GraphQL Schema设计法则
➠ Resolver嵌套查询优化
➠ DataLoader解决N+1问题
➠ 实战：社交网络关系API

挑战任务：
✍️ 尝试实现订单服务与通知服务的Saga事务，分享你的状态转换设计！

通过本期实战，你已掌握分布式系统核心能力：
✅ gRPC 跨服务高性能通信
✅ Kafka 分布式消息总线
✅ Saga 分布式事务模式
✅ 金融系统架构设计实践
✅ 分布式追踪与监控

架构前瞻：这些模式能力可扩展到每天处理百万级交易的支付系统、千亿级请求的社交平台等超大型分布式场景。

相关推荐

1. NestJS 从入门到精通 11：综合实战 - 构建知乎风格的全栈社区平台（下） 2025-06-24
2. NestJS 从入门到精通 10：综合实战 - 构建知乎风格的全栈社区平台（上） 2025-06-24
3. NestJS 从入门到精通 09：性能调优大师 - 百万并发下的极致优化策略 2025-06-24
4. NestJS 从入门到精通 08：安全架构实战 - 构建金融级应用的8重防护门 2025-06-24
5. NestJS 从入门到精通 07：配置管理与部署实战 - 多环境与容器化方案 2025-06-24