负载均衡策略

负载均衡策略的基本概念

负载均衡策略是分布式系统中用于分配工作负载的核心机制。在Node.js应用中，合理的负载均衡能有效提升系统吞吐量、降低响应时间并提高资源利用率。常见的负载均衡算法包括轮询、加权轮询、最少连接数、IP哈希等，每种策略适用于不同的业务场景。

轮询策略

轮询是最基础的负载均衡策略，将请求按顺序分配给后端服务器。Node.js中可以通过简单的数组索引实现：

const servers = ['server1', 'server2', 'server3'];
let currentIndex = 0;

function roundRobin() {
  const server = servers[currentIndex];
  currentIndex = (currentIndex + 1) % servers.length;
  return server;
}

// 使用示例
for (let i = 0; i < 5; i++) {
  console.log(roundRobin()); // 输出: server1, server2, server3, server1, server2
}

这种策略适合服务器配置相同的场景，但无法处理服务器性能差异的情况。

加权轮询策略

加权轮询在基础轮询上增加了权重概念，性能更强的服务器获得更多请求。实现时需要维护权重计数器：

const weightedServers = [
  { server: 'server1', weight: 3 },
  { server: 'server2', weight: 2 },
  { server: 'server3', weight: 1 }
];

let currentWeight = 0;
let gcdValue = 1; // 最大公约数
let maxWeight = 3;

function getGCD(a, b) {
  return b === 0 ? a : getGCD(b, a % b);
}

function weightedRoundRobin() {
  while (true) {
    currentWeight = (currentWeight + 1) % (maxWeight + 1);
    for (const { server, weight } of weightedServers) {
      if (weight >= currentWeight) {
        return server;
      }
    }
  }
}

最少连接数策略

最少连接数策略将新请求分配给当前连接数最少的服务器，适合处理长连接场景。实现需要实时监控服务器状态：

class LeastConnections {
  constructor(servers) {
    this.servers = servers.map(server => ({
      ...server,
      connections: 0
    }));
  }

  getServer() {
    let minConnections = Infinity;
    let selectedServer = null;

    for (const server of this.servers) {
      if (server.connections < minConnections) {
        minConnections = server.connections;
        selectedServer = server;
      }
    }

    if (selectedServer) {
      selectedServer.connections++;
      return selectedServer;
    }
    return null;
  }

  releaseServer(server) {
    const found = this.servers.find(s => s === server);
    if (found) found.connections--;
  }
}

IP哈希策略

IP哈希策略通过客户端IP计算哈希值确定目标服务器，保证同一客户端的请求总是落到同一服务器：

function ipHash(ip, serverCount) {
  let hash = 0;
  for (let i = 0; i < ip.length; i++) {
    hash = (hash << 5) - hash + ip.charCodeAt(i);
    hash |= 0; // 转换为32位整数
  }
  return Math.abs(hash) % serverCount;
}

const servers = ['server1', 'server2', 'server3'];
const clientIP = '192.168.1.100';

const selectedIndex = ipHash(clientIP, servers.length);
console.log(servers[selectedIndex]); // 输出根据IP哈希确定的服务器

动态权重调整

高级负载均衡系统会根据服务器实时性能动态调整权重。以下示例展示基于CPU使用率的动态调整：

class DynamicWeightBalancer {
  constructor(servers) {
    this.servers = servers;
    this.performanceHistory = {};
  }

  async updateServerMetrics() {
    for (const server of this.servers) {
      const metrics = await fetchServerMetrics(server.url);
      this.performanceHistory[server.id] = {
        cpu: metrics.cpu,
        memory: metrics.memory,
        responseTime: metrics.responseTime
      };
    }
  }

  calculateWeight(serverId) {
    const metrics = this.performanceHistory[serverId];
    if (!metrics) return 1;
    
    // 简单的权重计算公式
    const cpuFactor = 1 - (metrics.cpu / 100);
    const memoryFactor = 1 - (metrics.memory.used / metrics.memory.total);
    const responseFactor = 1 / (1 + Math.log(metrics.responseTime));
    
    return Math.max(1, Math.round((cpuFactor + memoryFactor + responseFactor) * 10));
  }

  async getServer() {
    await this.updateServerMetrics();
    const weightedServers = this.servers.map(server => ({
      server,
      weight: this.calculateWeight(server.id)
    }));
    
    // 使用加权轮询算法选择服务器
    return weightedRoundRobin(weightedServers);
  }
}

健康检查机制

完善的负载均衡需要包含健康检查机制，自动剔除故障节点：

class HealthCheck {
  constructor(servers) {
    this.servers = servers;
    this.healthyServers = [...servers];
    this.checkInterval = 30000; // 30秒检查一次
  }

  async checkServer(server) {
    try {
      const response = await fetch(`${server.url}/health`);
      return response.ok;
    } catch (error) {
      return false;
    }
  }

  async performCheck() {
    const checks = this.servers.map(async server => {
      const isHealthy = await this.checkServer(server);
      return { server, isHealthy };
    });

    const results = await Promise.all(checks);
    this.healthyServers = results
      .filter(({ isHealthy }) => isHealthy)
      .map(({ server }) => server);
  }

  start() {
    this.timer = setInterval(() => this.performCheck(), this.checkInterval);
    this.performCheck(); // 立即执行首次检查
  }

  stop() {
    clearInterval(this.timer);
  }
}

会话保持技术

某些应用需要保持用户会话，可通过cookie注入实现：

const cookie = require('cookie');

function sessionStickyMiddleware(balancer) {
  return async (req, res, next) => {
    const serverCookie = req.cookies?.server_id;
    
    if (serverCookie && balancer.hasServer(serverCookie)) {
      req.targetServer = serverCookie;
    } else {
      const server = await balancer.getServer();
      req.targetServer = server.id;
      res.setHeader('Set-Cookie', 
        cookie.serialize('server_id', server.id, { 
          httpOnly: true,
          maxAge: 60 * 60 * 24 // 1天
        })
      );
    }
    next();
  };
}

微服务架构中的负载均衡

在微服务架构中，通常结合服务发现实现动态负载均衡：

const Consul = require('consul');

class ServiceDiscoveryBalancer {
  constructor(serviceName) {
    this.serviceName = serviceName;
    this.consul = new Consul();
    this.services = [];
    this.watchInterval = null;
  }

  async discoverServices() {
    const result = await this.consul.health.service({
      service: this.serviceName,
      passing: true
    });
    this.services = result.map(entry => ({
      id: entry.Service.ID,
      address: entry.Service.Address,
      port: entry.Service.Port,
      tags: entry.Service.Tags
    }));
  }

  startWatching() {
    this.discoverServices();
    this.watchInterval = setInterval(
      () => this.discoverServices(), 
      10000 // 每10秒刷新一次服务列表
    );
  }

  getServer() {
    if (this.services.length === 0) {
      throw new Error('No available services');
    }
    // 使用最少连接数算法选择服务器
    return this.services.reduce((prev, curr) => 
      (prev.connections < curr.connections) ? prev : curr
    );
  }
}

云原生环境下的负载均衡

在Kubernetes环境中，Node.js应用可以充分利用Ingress控制器：

apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: nodejs-ingress
  annotations:
    nginx.ingress.kubernetes.io/affinity: "cookie"
    nginx.ingress.kubernetes.io/load-balance: "ewma"
spec:
  rules:
  - host: myapp.example.com
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: nodejs-service
            port:
              number: 3000

对应的Node.js服务需要正确处理健康检查端点：

app.get('/health', (req, res) => {
  const health = {
    status: 'UP',
    checks: [
      { name: 'database', status: checkDatabase() },
      { name: 'cache', status: checkCache() }
    ]
  };
  res.json(health);
});

function checkDatabase() {
  // 实现数据库健康检查逻辑
  return 'UP';
}

function checkCache() {
  // 实现缓存健康检查逻辑
  return 'UP';
}

性能监控与调优

实现负载均衡后需要持续监控性能指标：

const prometheus = require('prom-client');

const requestDuration = new prometheus.Histogram({
  name: 'http_request_duration_seconds',
  help: 'Duration of HTTP requests in seconds',
  labelNames: ['method', 'route', 'status', 'server'],
  buckets: [0.1, 0.5, 1, 2, 5]
});

function monitorMiddleware(req, res, next) {
  const start = Date.now();
  const originalEnd = res.end;

  res.end = function(...args) {
    const duration = (Date.now() - start) / 1000;
    requestDuration
      .labels(req.method, req.path, res.statusCode, req.targetServer)
      .observe(duration);
    originalEnd.apply(res, args);
  };

  next();
}

异常处理与熔断机制

负载均衡系统需要包含完善的异常处理：

class CircuitBreaker {
  constructor(server, threshold = 5, timeout = 30000) {
    this.server = server;
    this.failureCount = 0;
    this.threshold = threshold;
    this.timeout = timeout;
    this.state = 'CLOSED';
    this.nextAttempt = 0;
  }

  async execute(request) {
    if (this.state === 'OPEN') {
      if (Date.now() > this.nextAttempt) {
        this.state = 'HALF-OPEN';
      } else {
        throw new Error('Circuit breaker is open');
      }
    }

    try {
      const response = await request(this.server);
      if (this.state === 'HALF-OPEN') {
        this.reset();
      }
      return response;
    } catch (error) {
      this.failureCount++;
      if (this.failureCount >= this.threshold) {
        this.trip();
      }
      throw error;
    }
  }

  trip() {
    this.state = 'OPEN';
    this.nextAttempt = Date.now() + this.timeout;
  }

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