Node.js 微服务架构实践

微服务架构已经成为现代应用开发的主流模式。Node.js 凭借其轻量级、高性能的特性，成为构建微服务的理想选择。

微服务架构基础

微服务架构将大型应用拆分为多个小型、独立的服务，每个服务负责特定的业务功能。

微服务的优势：

服务发现与注册

在微服务架构中，服务发现是一个关键组件。

// 使用 Consul 进行服务注册
const consul = require('consul')();

class ServiceRegistry {
  async register(serviceName, port, health) {
    const serviceId = `${serviceName}-${port}`;

    await consul.agent.service.register({
      id: serviceId,
      name: serviceName,
      port: port,
      check: {
        http: `http://localhost:${port}${health}`,
        interval: '10s'
      }
    });

    console.log(`Service ${serviceName} registered`);
  }

  async discover(serviceName) {
    const services = await consul.health.service(serviceName);
    return services[0].map(service => ({
      host: service.Service.Address,
      port: service.Service.Port
    }));
  }
}

API 网关设计

API 网关作为微服务的统一入口，处理路由、认证、限流等横切关注点。

// 简单的 API 网关实现
const express = require('express');
const httpProxy = require('http-proxy-middleware');

class APIGateway {
  constructor() {
    this.app = express();
    this.setupMiddleware();
    this.setupRoutes();
  }

  setupMiddleware() {
    // 认证中间件
    this.app.use('/api', this.authenticate);

    // 限流中间件
    this.app.use('/api', this.rateLimit);
  }

  setupRoutes() {
    // 用户服务路由
    this.app.use('/api/users', httpProxy({
      target: 'http://user-service:3001',
      changeOrigin: true,
      pathRewrite: {
        '^/api/users': ''
      }
    }));

    // 订单服务路由
    this.app.use('/api/orders', httpProxy({
      target: 'http://order-service:3002',
      changeOrigin: true,
      pathRewrite: {
        '^/api/orders': ''
      }
    }));
  }

  authenticate(req, res, next) {
    // JWT 认证逻辑
    const token = req.headers.authorization;
    if (!token) {
      return res.status(401).json({ error: 'Unauthorized' });
    }
    // 验证 token...
    next();
  }

  rateLimit(req, res, next) {
    // 限流逻辑
    next();
  }
}

服务间通信

微服务之间需要高效的通信机制。

HTTP/REST 通信

// HTTP 客户端封装
class ServiceClient {
  constructor(baseURL) {
    this.baseURL = baseURL;
    this.timeout = 5000;
  }

  async get(path, options = {}) {
    const response = await fetch(`${this.baseURL}${path}`, {
      method: 'GET',
      timeout: this.timeout,
      ...options
    });

    if (!response.ok) {
      throw new Error(`HTTP ${response.status}: ${response.statusText}`);
    }

    return response.json();
  }

  async post(path, data, options = {}) {
    const response = await fetch(`${this.baseURL}${path}`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        ...options.headers
      },
      body: JSON.stringify(data),
      timeout: this.timeout,
      ...options
    });

    if (!response.ok) {
      throw new Error(`HTTP ${response.status}: ${response.statusText}`);
    }

    return response.json();
  }
}

消息队列通信

// 使用 RabbitMQ 进行异步通信
const amqp = require('amqplib');

class MessageBroker {
  constructor() {
    this.connection = null;
    this.channel = null;
  }

  async connect() {
    this.connection = await amqp.connect('amqp://localhost');
    this.channel = await this.connection.createChannel();
  }

  async publish(exchange, routingKey, message) {
    await this.channel.assertExchange(exchange, 'topic', { durable: true });

    this.channel.publish(
      exchange,
      routingKey,
      Buffer.from(JSON.stringify(message)),
      { persistent: true }
    );
  }

  async subscribe(exchange, routingKey, callback) {
    await this.channel.assertExchange(exchange, 'topic', { durable: true });

    const queue = await this.channel.assertQueue('', { exclusive: true });
    await this.channel.bindQueue(queue.queue, exchange, routingKey);

    this.channel.consume(queue.queue, (msg) => {
      if (msg) {
        const content = JSON.parse(msg.content.toString());
        callback(content);
        this.channel.ack(msg);
      }
    });
  }
}

容错与熔断

在分布式系统中，容错机制至关重要。

// 熔断器模式实现
class CircuitBreaker {
  constructor(threshold = 5, timeout = 60000) {
    this.threshold = threshold;
    this.timeout = timeout;
    this.failureCount = 0;
    this.lastFailureTime = null;
    this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
  }

  async call(fn) {
    if (this.state === 'OPEN') {
      if (Date.now() - this.lastFailureTime > this.timeout) {
        this.state = 'HALF_OPEN';
      } else {
        throw new Error('Circuit breaker is OPEN');
      }
    }

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }

  onSuccess() {
    this.failureCount = 0;
    this.state = 'CLOSED';
  }

  onFailure() {
    this.failureCount++;
    this.lastFailureTime = Date.now();

    if (this.failureCount >= this.threshold) {
      this.state = 'OPEN';
    }
  }
}

监控与日志

微服务架构需要完善的监控和日志系统。

// 结构化日志
const winston = require('winston');

const logger = winston.createLogger({
  format: winston.format.combine(
    winston.format.timestamp(),
    winston.format.errors({ stack: true }),
    winston.format.json()
  ),
  transports: [
    new winston.transports.File({ filename: 'error.log', level: 'error' }),
    new winston.transports.File({ filename: 'combined.log' }),
    new winston.transports.Console({
      format: winston.format.simple()
    })
  ]
});

// 请求追踪中间件
function requestTracing(req, res, next) {
  const traceId = req.headers['x-trace-id'] || generateTraceId();
  req.traceId = traceId;
  res.setHeader('x-trace-id', traceId);

  logger.info('Request started', {
    traceId,
    method: req.method,
    url: req.url,
    userAgent: req.headers['user-agent']
  });

  next();
}

部署与容器化

使用 Docker 容器化微服务：

Dockerfile
FROM node:16-alpine

WORKDIR /app

COPY package*.json ./
RUN npm ci --only=production

COPY . .

EXPOSE 3000

USER node

CMD ["node", "index.js"]

总结

Node.js 微服务架构为构建可扩展、可维护的分布式系统提供了强大的基础。关键是要合理设计服务边界，选择合适的通信方式，并建立完善的监控和容错机制。

在实际项目中，建议从单体应用开始，随着业务复杂度的增加逐步拆分为微服务。记住，微服务不是银弹，需要根据具体业务需求来决定是否采用。