Node.js 微服务架构实战：服务拆分、通信、治理与分布式事务

// NestJS REST 控制器示例
@Controller('orders')
export class OrderController {
  constructor(private readonly orderService: OrderService) {}

  @Post()
  async createOrder(@Body() dto: CreateOrderDto): Promise<Order> {
    return this.orderService.create(dto);
  }

  @Get(':id')
  async getOrder(@Param('id') id: string): Promise<Order> {
    return this.orderService.findById(id);
  }
}

// order.proto —— gRPC 服务定义
syntax = "proto3";

package ecommerce;

service OrderService {
  rpc CreateOrder (CreateOrderRequest) returns (Order);
  rpc GetOrder (GetOrderRequest) returns (Order);
  rpc StreamOrderUpdates (StreamOrderRequest) returns (stream OrderUpdate);
}

message CreateOrderRequest {
  string user_id = 1;
  repeated OrderItem items = 2;
  string currency = 3;
}

message OrderItem {
  string product_id = 1;
  int32 quantity = 2;
  int64 unit_price_cents = 3;
}

message Order {
  string order_id = 1;
  string user_id = 2;
  OrderStatus status = 3;
  int64 total_cents = 4;
  repeated OrderItem items = 5;
  string created_at = 6;
}

enum OrderStatus {
  PENDING = 0;
  PAID = 1;
  SHIPPED = 2;
  DELIVERED = 3;
  CANCELLED = 4;
}

message GetOrderRequest {
  string order_id = 1;
}

message StreamOrderRequest {
  string order_id = 1;
}

message OrderUpdate {
  string order_id = 1;
  OrderStatus new_status = 2;
  string timestamp = 3;
}

// consul-service-registrar.js
const Consul = require('consul');

const consul = new Consul({ host: 'consul-server', port: 8500 });
const serviceId = `order-service-${process.env.HOSTNAME || Date.now()}`;

async function register() {
  await consul.agent.service.register({
    id: serviceId,
    name: 'order-service',
    tags: ['nodejs', 'v1'],
    port: 3000,
    check: {
      http: 'http://localhost:3000/health',
      interval: '10s',
      timeout: '5s',
      deregistercriticalserviceafter: '30s',
    },
  });
  console.log(`Registered ${serviceId} with Consul`);
}

async function deregister(signal) {
  await consul.agent.service.deregister(serviceId);
  console.log(`Deregistered ${serviceId} (${signal})`);
  process.exit(0);
}

process.on('SIGINT', () => deregister('SIGINT'));
process.on('SIGTERM', () => deregister('SIGTERM'));

register();

apiVersion: v1
kind: Service
metadata:
  name: order-service
spec:
  selector:
    app: order-service
  ports:
    - port: 80
      targetPort: 3000

# kong-declarative.yml
_format_version: "3.0"
services:
  - name: order-service
    url: http://order-service:3000
    routes:
      - name: order-routes
        paths:
          - /orders
    plugins:
      - name: rate-limiting
        config:
          minute: 100
          policy: redis
          redis_host: redis-cluster
      - name: jwt
        config:
          uri_param_names: []
          cookie_names: []
          key_claim_name: iss
          secret_is_base64: false
          claims_to_verify:
            - exp

// Web BFF —— 聚合首页数据的 NestJS 服务
@Controller('home')
export class HomeBffController {
  constructor(
    private userClient: UserGrpcClient,
    private productClient: ProductGrpcClient,
    private promoClient: PromoGrpcClient,
  ) {}

  @Get('dashboard')
  async getDashboard(@Req() req: Request) {
    const userId = req.user.sub;

    // 并行调用多个下游服务
    const [profile, recommendations, coupons] = await Promise.all([
      this.userClient.getProfile(userId),
      this.productClient.getRecommendations(userId),
      this.promoClient.getActiveCoupons(userId),
    ]);

    // 聚合为前端需要的结构
    return {
      greeting: `欢迎回来，${profile.name}`,
      defaultAddress: profile.addresses[0],
      recommendations: recommendations.map(r => ({
        id: r.id,
        title: r.title,
        price: this.formatPrice(r.price),
        image: r.images[0],
        inStock: r.stock > 0,
      })),
      coupons: coupons.filter(c => !c.expired).map(c => ({
        code: c.code,
        description: c.description,
        discount: `${c.discountPercent}%`,
      })),
    };
  }

  private formatPrice(cents: number): string {
    return `¥${(cents / 100).toFixed(2)}`;
  }
}

// 基于 opossum 的熔断器封装
import CircuitBreaker from 'opossum';

const options = {
  timeout: 3000,           // 3 秒超时
  errorThresholdPercentage: 50,  // 失败率 50% 触发熔断
  resetTimeout: 30000,     // 30 秒后尝试恢复
  volumeThreshold: 10,     // 至少 10 次请求才计算失败率
};

const breaker = new CircuitBreaker(
  (paymentData) => paymentService.charge(paymentData),
  options
);

breaker.on('open', () => console.warn('Payment circuit opened'));
breaker.on('halfOpen', () => console.warn('Payment circuit half-open'));
breaker.on('close', () => console.info('Payment circuit closed'));

// 使用
async function processPayment(data: PaymentData) {
  try {
    return await breaker.fire(data);
  } catch (err) {
    if (breaker.opened) {
      // 熔断中：返回降级响应或排队稍后处理
      return { status: 'queued', message: '支付服务暂时不可用，已加入队列' };
    }
    throw err;
  }
}

import { backOff } from 'exponential-backoff';

async function reliableFetch(url: string) {
  return backOff(
    () => fetch(url),
    {
      numOfAttempts: 5,
      startingDelay: 100,
      maxDelay: 5000,
      timeMultiple: 2,  // 指数退避：100ms → 200ms → 400ms → ...
      jitter: 'full',   // 随机抖动避免惊群效应
      retry: (err, attempt) => {
        // 仅对可重试错误重试
        return err.status >= 500 || err.code === 'ECONNRESET';
      },
    }
  );
}

// Axios 请求超时配置
const client = axios.create({
  timeout: 2000,           // 2 秒请求超时
  timeoutErrorMessage: 'Service response timeout',
});

// 级联超时策略：网关层 5s → BFF 层 4s → 服务层 3s → 数据库层 2s

// tracing.ts —— NestJS 项目的 OpenTelemetry 初始化
import { NodeSDK } from '@opentelemetry/sdk-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-grpc';
import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
import { Resource } from '@opentelemetry/resources';
import { SemanticResourceAttributes } from '@opentelemetry/semantic-conventions';

const sdk = new NodeSDK({
  resource: new Resource({
    [SemanticResourceAttributes.SERVICE_NAME]: 'order-service',
    [SemanticResourceAttributes.SERVICE_VERSION]: '1.0.0',
    [SemanticResourceAttributes.DEPLOYMENT_ENVIRONMENT]: 'production',
  }),
  traceExporter: new OTLPTraceExporter({
    url: process.env.JAEGER_COLLECTOR_URL || 'http://jaeger:4317',
  }),
  instrumentations: [
    getNodeAutoInstrumentations({
      '@opentelemetry/instrumentation-http': { enabled: true },
      '@opentelemetry/instrumentation-grpc': { enabled: true },
    }),
  ],
});

sdk.start();

process.on('SIGTERM', () => sdk.shutdown());

import { trace } from '@opentelemetry/api';

const tracer = trace.getTracer('order-service');

async function createOrder(dto: CreateOrderDto) {
  return tracer.startActiveSpan('order.create', {
    attributes: { 'order.user_id': dto.userId },
  }, async (span) => {
    try {
      const order = await db.orders.insert(dto);
      span.setAttribute('order.id', order.id);
      return order;
    } catch (err) {
      span.recordException(err);
      throw err;
    } finally {
      span.end();
    }
  });
}

// HTTP 请求中注入 trace context（自动，库已处理）
// 手动注入 gRPC metadata
import { propagation, context } from '@opentelemetry/api';
import { Metadata } from '@grpc/grpc-js';

function injectTraceContext(meta: Metadata): Metadata {
  const carrier = {};
  propagation.inject(context.active(), carrier);
  for (const [key, value] of Object.entries(carrier)) {
    meta.add(key, String(value));
  }
  return meta;
}

// NestJS Saga 编排器示例
@Injectable()
export class CreateOrderSaga {
  constructor(
    private inventoryClient: InventoryGrpcClient,
    private paymentClient: PaymentGrpcClient,
    private orderClient: OrderGrpcClient,
    private sagaLog: SagaLogRepository,
  ) {}

  async execute(dto: CreateOrderDto): Promise<string> {
    const sagaId = generateUUID();
    const compensationStack: Compensation[] = [];

    try {
      // Step 1: 扣减库存
      const reservation = await this.inventoryClient.reserveStock({
        productId: dto.productId,
        quantity: dto.quantity,
        sagaId,
      });
      compensationStack.push({
        action: () => this.inventoryClient.releaseReservation(reservation.id),
      });
      await this.sagaLog.recordStep(sagaId, 'inventory_reserved', reservation);

      // Step 2: 支付授权
      const payment = await this.paymentClient.authorize({
        userId: dto.userId,
        amount: dto.amount,
        sagaId,
      });
      compensationStack.push({
        action: () => this.paymentClient.voidAuthorization(payment.id),
      });
      await this.sagaLog.recordStep(sagaId, 'payment_authorized', payment);

      // Step 3: 创建订单
      const order = await this.orderClient.createOrder({
        ...dto,
        reservationId: reservation.id,
        paymentId: payment.id,
        sagaId,
      });
      await this.sagaLog.recordStep(sagaId, 'order_created', order);

      return order.id;

    } catch (err) {
      // 执行补偿（LIFO 顺序）
      while (compensationStack.length > 0) {
        const comp = compensationStack.pop();
        try {
          await comp.action();
        } catch (compErr) {
          // 补偿失败 → 记录待人工介入
          await this.sagaLog.recordCompensationFailure(sagaId, compErr);
        }
      }
      throw new OrderCreationFailedException(sagaId, err.message);
    }
  }
}

// proto/inventory.proto
syntax = "proto3";

package inventory;

service InventoryService {
  rpc CheckStock (CheckStockRequest) returns (StockStatus);
  rpc ReserveStock (ReserveRequest) returns (Reservation);
}

message CheckStockRequest {
  string product_id = 1;
  int32 quantity = 2;
}

message StockStatus {
  string product_id = 1;
  int32 available = 2;
  bool sufficient = 3;
}

message ReserveRequest {
  string product_id = 1;
  int32 quantity = 2;
  string order_id = 3;
}

message Reservation {
  string reservation_id = 1;
  string product_id = 2;
  int32 reserved_quantity = 3;
  int64 expires_at = 4;
}

// inventory-service/src/inventory.controller.ts
import { Controller } from '@nestjs/common';
import { GrpcMethod } from '@nestjs/microservices';
import { InventoryService } from './inventory.service';

interface CheckStockRequest {
  productId: string;
  quantity: number;
}

interface ReserveRequest {
  productId: string;
  quantity: number;
  orderId: string;
}

@Controller()
export class InventoryController {
  constructor(private readonly inventoryService: InventoryService) {}

  @GrpcMethod('InventoryService', 'CheckStock')
  async checkStock(data: CheckStockRequest) {
    return this.inventoryService.checkAvailability(
      data.productId,
      data.quantity,
    );
  }

  @GrpcMethod('InventoryService', 'ReserveStock')
  async reserveStock(data: ReserveRequest) {
    return this.inventoryService.reserve(
      data.productId,
      data.quantity,
      data.orderId,
    );
  }
}

// inventory-service/src/main.ts
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { join } from 'path';
import { InventoryModule } from './app.module';

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

  await app.listen();
  console.log('Inventory gRPC Microservice running on port 50051');
}

bootstrap();

// order-service/src/app.module.ts
import { Module } from '@nestjs/common';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { join } from 'path';
import { OrderController } from './order.controller';
import { OrderService } from './order.service';

@Module({
  imports: [
    ClientsModule.register([
      {
        name: 'INVENTORY_PACKAGE',
        transport: Transport.GRPC,
        options: {
          package: 'inventory',
          protoPath: join(__dirname, '../proto/inventory.proto'),
          url: process.env.INVENTORY_SERVICE_URL || 'localhost:50051',
        },
      },
    ]),
  ],
  controllers: [OrderController],
  providers: [OrderService],
})
export class OrderModule {}

// order-service/src/order.service.ts
import { Inject, Injectable, OnModuleInit } from '@nestjs/common';
import { ClientGrpc } from '@nestjs/microservices';
import { lastValueFrom } from 'rxjs';

interface InventoryService {
  checkStock(data: { productId: string; quantity: number }): Promise<any>;
  reserveStock(data: {
    productId: string;
    quantity: number;
    orderId: string;
  }): Promise<any>;
}

@Injectable()
export class OrderService implements OnModuleInit {
  private inventoryService: InventoryService;

  constructor(
    @Inject('INVENTORY_PACKAGE') private client: ClientGrpc,
  ) {}

  onModuleInit() {
    this.inventoryService = this.client.getService<InventoryService>('InventoryService');
  }

  async createOrder(productId: string, quantity: number, userId: string) {
    // Step 1: 检查库存
    const stock = await lastValueFrom(
      this.inventoryService.checkStock({ productId, quantity }),
    );

    if (!stock.sufficient) {
      throw new Error(`Insufficient stock for product ${productId}`);
    }

    // Step 2: 预留库存
    const reservation = await lastValueFrom(
      this.inventoryService.reserveStock({
        productId,
        quantity,
        orderId: `order_${Date.now()}`,
      }),
    );

    // Step 3: 创建订单记录（省略实际实现）
    return {
      orderId: reservation.orderId,
      status: 'created',
      reservationId: reservation.reservationId,
    };
  }
}

// order-service/src/order.controller.ts
import { Body, Controller, Post } from '@nestjs/common';
import { OrderService } from './order.service';

class CreateOrderDto {
  productId: string;
  quantity: number;
  userId: string;
}

@Controller('orders')
export class OrderController {
  constructor(private readonly orderService: OrderService) {}

  @Post()
  async createOrder(@Body() dto: CreateOrderDto) {
    return this.orderService.createOrder(
      dto.productId,
      dto.quantity,
      dto.userId,
    );
  }
}

version: '3.8'

services:
  inventory-service:
    build: ./inventory-service
    ports:
      - "50051:50051"
    environment:
      - NODE_ENV=production
    healthcheck:
      test: ["CMD", "grpc-health-probe", "-addr=:50051"]
      interval: 10s
      timeout: 5s
      retries: 3

  order-service:
    build: ./order-service
    ports:
      - "3000:3000"
    environment:
      - INVENTORY_SERVICE_URL=inventory-service:50051
      - NODE_ENV=production
    depends_on:
      inventory-service:
        condition: service_healthy