設計負載均衡器 — Round Robin、一致性雜湊與健康檢查實戰 | 資料結構與演算法

// ================================================================
// Round Robin — 最簡單的負載均衡
// ================================================================

interface ServerNode {
  id: string;
  host: string;
  port: number;
  weight: number; // 1~10，用於加權演算法
}

class RoundRobinBalancer {
  private index = 0;

  constructor(private servers: ServerNode[]) {}

  next(): ServerNode | null {
    if (this.servers.length === 0) return null;
    const node = this.servers[this.index % this.servers.length];
    this.index++;
    return node;
  }

  // 動態新增/移除節點
  addServer(server: ServerNode): void {
    this.servers.push(server);
  }

  removeServer(serverId: string): void {
    this.servers = this.servers.filter((s) => s.id !== serverId);
    this.index = this.index % Math.max(1, this.servers.length);
  }
}

// 使用範例
const servers: ServerNode[] = [
  { id: "server-1", host: "10.0.0.1", port: 8080, weight: 2 },
  { id: "server-2", host: "10.0.0.2", port: 8080, weight: 4 },
  { id: "server-3", host: "10.0.0.3", port: 8080, weight: 2 },
];

const rr = new RoundRobinBalancer(servers);
for (let i = 0; i < 6; i++) {
  console.log(`請求 #${i + 1} → ${rr.next()?.id}`);
}
// 輸出：
// 請求 #1 → server-1
// 請求 #2 → server-2
// 請求 #3 → server-3
// 請求 #4 → server-1
// 請求 #5 → server-2
// 請求 #6 → server-3

// ================================================================
// Weighted Round Robin — Nginx 使用的平滑加權輪詢
// ================================================================

class WeightedRoundRobinBalancer {
  private currentWeights: number[];

  constructor(private readonly servers: ServerNode[]) {
    this.currentWeights = new Array(servers.length).fill(0);
  }

  next(): ServerNode | null {
    if (this.servers.length === 0) return null;

    const totalWeight = this.servers.reduce((sum, s) => sum + s.weight, 0);

    // 每輪：所有節點 current_weight += weight
    for (let i = 0; i < this.servers.length; i++) {
      this.currentWeights[i] += this.servers[i].weight;
    }

    // 選 current_weight 最大的節點
    let maxIdx = 0;
    for (let i = 1; i < this.servers.length; i++) {
      if (this.currentWeights[i] > this.currentWeights[maxIdx]) {
        maxIdx = i;
      }
    }

    // 選中後減去 total_weight
    this.currentWeights[maxIdx] -= totalWeight;
    return this.servers[maxIdx];
  }
}

// 使用範例：驗證分配比例
const wrr = new WeightedRoundRobinBalancer(servers);
const counts: Record<string, number> = {};
for (let i = 0; i < 800; i++) {
  const node = wrr.next()!;
  counts[node.id] = (counts[node.id] ?? 0) + 1;
}
console.log(counts);
// 輸出：{ 'server-1': 200, 'server-2': 400, 'server-3': 200 }
// → server-2（weight=4）得到 50%，符合權重比例

// ================================================================
// Least Connections — 即時負載感知
// ================================================================

class LeastConnectionsBalancer {
  private connections: Map<string, number>;

  constructor(private servers: ServerNode[]) {
    this.connections = new Map(servers.map((s) => [s.id, 0]));
  }

  // 獲取下一個節點（連線數 +1）
  acquire(): ServerNode | null {
    if (this.servers.length === 0) return null;

    let minConn = Infinity;
    let chosen = this.servers[0];

    for (const server of this.servers) {
      const conn = this.connections.get(server.id) ?? 0;
      if (conn < minConn) {
        minConn = conn;
        chosen = server;
      }
    }

    // 增加連線計數
    this.connections.set(
      chosen.id,
      (this.connections.get(chosen.id) ?? 0) + 1
    );
    return chosen;
  }

  // 請求完成後釋放連線
  release(serverId: string): void {
    const current = this.connections.get(serverId) ?? 0;
    this.connections.set(serverId, Math.max(0, current - 1));
  }

  getConnections(): Map<string, number> {
    return new Map(this.connections);
  }
}

// 使用範例：模擬長短請求混合場景
const lc = new LeastConnectionsBalancer(servers);

const s1 = lc.acquire(); // server-1（連線數 0）
const s2 = lc.acquire(); // server-2（連線數 0）
const s3 = lc.acquire(); // server-3（連線數 0）
console.log("分配後：", Object.fromEntries(lc.getConnections()));
// 輸出：{ 'server-1': 1, 'server-2': 1, 'server-3': 1 }

lc.release(s1!.id); // server-1 釋放連線
const s4 = lc.acquire(); // server-1（連線數最少 = 0）
console.log("server-1 釋放後再分配：", s4?.id);
// 輸出：server-1

// ================================================================
// IP Hash — Session Affinity（同 IP 固定路由）
// ================================================================

class IPHashBalancer {
  constructor(private servers: ServerNode[]) {}

  next(clientIP: string): ServerNode | null {
    if (this.servers.length === 0) return null;
    const hash = this.simpleHash(clientIP);
    return this.servers[hash % this.servers.length];
  }

  private simpleHash(str: string): number {
    let hash = 0;
    for (let i = 0; i < str.length; i++) {
      hash = (hash * 31 + str.charCodeAt(i)) >>> 0; // 保持為 uint32
    }
    return hash;
  }
}

// 使用範例：驗證同 IP 固定路由
const ipHash = new IPHashBalancer(servers);
console.log(ipHash.next("192.168.1.100")?.id); // 固定結果
console.log(ipHash.next("192.168.1.100")?.id); // 同 IP → 同節點
console.log(ipHash.next("10.0.0.55")?.id); // 不同 IP → 可能不同節點

// ================================================================
// Consistent Hash Ring — 完整實作
// 支援：虛擬節點、加權分配、動態增刪節點、O(log V) 路由
// ================================================================
import * as crypto from "crypto";

class ConsistentHashRing {
  private readonly ring: Map<number, ServerNode> = new Map();
  private readonly sortedKeys: number[] = [];
  private readonly nodes: Map<string, ServerNode> = new Map();
  private readonly virtualNodesPerWeight: number;

  constructor(virtualNodesPerWeight: number = 150) {
    this.virtualNodesPerWeight = virtualNodesPerWeight;
  }

  // 新增節點：根據 weight 決定虛擬節點數量
  addNode(node: ServerNode): void {
    if (this.nodes.has(node.id)) {
      this.removeNode(node.id); // 先移除再重新加入（冪等操作）
    }
    this.nodes.set(node.id, node);

    const virtualCount = this.virtualNodesPerWeight * node.weight;
    for (let i = 0; i < virtualCount; i++) {
      const hashKey = this.hash(`${node.id}#${i}`);
      this.ring.set(hashKey, node);
      this.insertSorted(hashKey);
    }
  }

  // 移除節點：從環上移除所有虛擬節點
  removeNode(nodeId: string): void {
    const node = this.nodes.get(nodeId);
    if (!node) return;

    this.nodes.delete(nodeId);
    const virtualCount = this.virtualNodesPerWeight * node.weight;
    for (let i = 0; i < virtualCount; i++) {
      const hashKey = this.hash(`${nodeId}#${i}`);
      this.ring.delete(hashKey);
      this.removeSorted(hashKey);
    }
  }

  // 路由：O(log V) 二分搜尋找順時針最近節點
  getNode(key: string): ServerNode | null {
    if (this.sortedKeys.length === 0) return null;

    const hashKey = this.hash(key);
    const idx = this.binarySearchCeil(hashKey);

    // 若 hashKey 大於所有點，回到環的起點（第一個節點）
    const ringIdx = idx < this.sortedKeys.length ? idx : 0;
    const targetHash = this.sortedKeys[ringIdx];
    return this.ring.get(targetHash) ?? null;
  }

  // 取得 key 對應的 N 個不重複節點（用於資料複製/備援）
  getNodes(key: string, count: number): ServerNode[] {
    if (count >= this.nodes.size) {
      return [...this.nodes.values()];
    }

    const hashKey = this.hash(key);
    let idx = this.binarySearchCeil(hashKey);
    if (idx >= this.sortedKeys.length) idx = 0;

    const result: ServerNode[] = [];
    const seenIds = new Set<string>();

    while (result.length < count) {
      const node = this.ring.get(this.sortedKeys[idx])!;
      if (!seenIds.has(node.id)) {
        seenIds.add(node.id);
        result.push(node);
      }
      idx = (idx + 1) % this.sortedKeys.length;
    }

    return result;
  }

  // 統計：各節點的虛擬節點分布
  getDistribution(): Map<string, number> {
    const dist = new Map<string, number>();
    for (const node of this.ring.values()) {
      dist.set(node.id, (dist.get(node.id) ?? 0) + 1);
    }
    return dist;
  }

  get nodeCount(): number {
    return this.nodes.size;
  }

  get virtualNodeCount(): number {
    return this.sortedKeys.length;
  }

  // ---- 私有工具函數 ----

  private hash(key: string): number {
    const buf = crypto.createHash("md5").update(key).digest();
    return buf.readUInt32BE(0); // 取前 4 bytes 作為 uint32
  }

  private insertSorted(key: number): void {
    let lo = 0,
      hi = this.sortedKeys.length;
    while (lo < hi) {
      const mid = (lo + hi) >> 1;
      if (this.sortedKeys[mid] < key) lo = mid + 1;
      else hi = mid;
    }
    this.sortedKeys.splice(lo, 0, key);
  }

  private removeSorted(key: number): void {
    const idx = this.binarySearchExact(key);
    if (idx !== -1) this.sortedKeys.splice(idx, 1);
  }

  private binarySearchCeil(target: number): number {
    let lo = 0,
      hi = this.sortedKeys.length;
    while (lo < hi) {
      const mid = (lo + hi) >> 1;
      if (this.sortedKeys[mid] < target) lo = mid + 1;
      else hi = mid;
    }
    return lo;
  }

  private binarySearchExact(target: number): number {
    let lo = 0,
      hi = this.sortedKeys.length - 1;
    while (lo <= hi) {
      const mid = (lo + hi) >> 1;
      if (this.sortedKeys[mid] === target) return mid;
      if (this.sortedKeys[mid] < target) lo = mid + 1;
      else hi = mid - 1;
    }
    return -1;
  }
}

// 使用範例
const ring = new ConsistentHashRing(150);
servers.forEach((s) => ring.addNode(s));

console.log("虛擬節點總數：", ring.virtualNodeCount);
// 輸出：1200（(2+4+2)×150）

console.log("分布：", Object.fromEntries(ring.getDistribution()));
// 輸出：{ 'server-1': 300, 'server-2': 600, 'server-3': 300 }

// 一致性路由測試
const testKeys = ["user-12345", "session-abc", "cache-key-99"];
for (const key of testKeys) {
  console.log(`${key} → ${ring.getNode(key)?.id}`);
}

// 節點故障模擬
console.log("\n--- 移除 server-2 ---");
ring.removeNode("server-2");
for (const key of testKeys) {
  console.log(`${key} → ${ring.getNode(key)?.id}`);
}
// → 原本路由到 server-2 的 key 會被重新分配到 server-1 或 server-3
// → 原本路由到 server-1/3 的 key 不受影響

// ================================================================
// Health Checker — 主動健康檢查與自動故障轉移
// ================================================================

interface HealthCheckConfig {
  intervalMs: number; // 檢查間隔（毫秒）
  timeoutMs: number; // 超時時間
  path: string; // 健康檢查路徑（如 "/health"）
  unhealthyThreshold: number; // 連續失敗幾次才標記為不健康
  healthyThreshold: number; // 連續成功幾次才標記為健康
}

class HealthChecker {
  private failureCount: Map<string, number> = new Map();
  private successCount: Map<string, number> = new Map();
  private healthyNodes: Set<string>;

  constructor(
    private readonly ring: ConsistentHashRing,
    private readonly allNodes: ServerNode[],
    private readonly config: HealthCheckConfig
  ) {
    this.healthyNodes = new Set(allNodes.map((n) => n.id));
  }

  // 啟動定時健康檢查
  start(): NodeJS.Timeout {
    return setInterval(async () => {
      await Promise.all(this.allNodes.map((node) => this.checkNode(node)));
    }, this.config.intervalMs);
  }

  private async checkNode(node: ServerNode): Promise<void> {
    const url = `http://${node.host}:${node.port}${this.config.path}`;
    try {
      const controller = new AbortController();
      const timeoutId = setTimeout(
        () => controller.abort(),
        this.config.timeoutMs
      );
      const response = await fetch(url, { signal: controller.signal });
      clearTimeout(timeoutId);

      if (response.ok) {
        this.onSuccess(node);
      } else {
        this.onFailure(node);
      }
    } catch {
      this.onFailure(node);
    }
  }

  private onSuccess(node: ServerNode): void {
    this.failureCount.set(node.id, 0);
    const successes = (this.successCount.get(node.id) ?? 0) + 1;
    this.successCount.set(node.id, successes);

    // 節點恢復健康：連續成功次數達到閾值
    if (
      !this.healthyNodes.has(node.id) &&
      successes >= this.config.healthyThreshold
    ) {
      this.healthyNodes.add(node.id);
      this.ring.addNode(node);
      console.log(`[HealthCheck] ✓ ${node.id} is back online`);
    }
  }

  private onFailure(node: ServerNode): void {
    this.successCount.set(node.id, 0);
    const failures = (this.failureCount.get(node.id) ?? 0) + 1;
    this.failureCount.set(node.id, failures);

    // 節點標記為不健康：連續失敗次數達到閾值
    if (
      this.healthyNodes.has(node.id) &&
      failures >= this.config.unhealthyThreshold
    ) {
      this.healthyNodes.delete(node.id);
      this.ring.removeNode(node.id);
      console.log(
        `[HealthCheck] ✗ ${node.id} is DOWN (${failures} consecutive failures)`
      );
    }
  }

  getHealthyNodes(): string[] {
    return [...this.healthyNodes];
  }

  isHealthy(nodeId: string): boolean {
    return this.healthyNodes.has(nodeId);
  }
}

// 使用範例
const healthConfig: HealthCheckConfig = {
  intervalMs: 5000, // 每 5 秒檢查一次
  timeoutMs: 2000, // 超時 2 秒視為失敗
  path: "/health",
  unhealthyThreshold: 3, // 連續 3 次失敗 → 標記不健康
  healthyThreshold: 2, // 連續 2 次成功 → 恢復健康
};

const healthyRing = new ConsistentHashRing(150);
servers.forEach((s) => healthyRing.addNode(s));
const checker = new HealthChecker(healthyRing, servers, healthConfig);

// checker.start();  // 啟動定時健康檢查
console.log("健康節點：", checker.getHealthyNodes());
// 輸出：['server-1', 'server-2', 'server-3']

// ================================================================
// 負載均衡器 — C++ 高效能完整實作
// 包含：Consistent Hash Ring、Weighted Round Robin、Least Connections
// ================================================================
#include <iostream>
#include <map>
#include <unordered_map>
#include <unordered_set>
#include <vector>
#include <string>
#include <atomic>
#include <mutex>
#include <optional>
#include <chrono>

struct ServerNode {
    std::string id;
    std::string host;
    int port;
    int weight;  // 1-10
};

// ================================================================
// 1. Consistent Hash Ring（一致性雜湊環）
// ================================================================
class ConsistentHashRing {
public:
    explicit ConsistentHashRing(int virtualNodesPerWeight = 150)
        : virtualNodesPerWeight_(virtualNodesPerWeight) {}

    void addNode(const ServerNode& node) {
        removeNode(node.id);  // 冪等操作
        nodes_[node.id] = node;

        const int virtualCount = virtualNodesPerWeight_ * node.weight;
        for (int i = 0; i < virtualCount; i++) {
            const std::string key = node.id + "#" + std::to_string(i);
            const uint32_t hash = hashKey(key);
            ring_[hash] = &nodes_[node.id];
        }
    }

    void removeNode(const std::string& nodeId) {
        auto it = nodes_.find(nodeId);
        if (it == nodes_.end()) return;

        const int virtualCount = virtualNodesPerWeight_ * it->second.weight;
        for (int i = 0; i < virtualCount; i++) {
            const std::string key = nodeId + "#" + std::to_string(i);
            ring_.erase(hashKey(key));
        }
        nodes_.erase(it);
    }

    // 路由：O(log V)，V = 虛擬節點總數
    std::optional<ServerNode> getNode(const std::string& requestKey) const {
        if (ring_.empty()) return std::nullopt;

        const uint32_t hash = hashKey(requestKey);
        auto it = ring_.lower_bound(hash);  // 找到第一個 >= hash 的位置

        if (it == ring_.end()) it = ring_.begin();  // 環繞
        return *it->second;
    }

    // 取得 N 個不重複節點（用於資料複製）
    std::vector<ServerNode> getNodes(
        const std::string& requestKey, int count
    ) const {
        if (ring_.empty()) return {};
        count = std::min(count, static_cast<int>(nodes_.size()));

        const uint32_t hash = hashKey(requestKey);
        auto it = ring_.lower_bound(hash);
        if (it == ring_.end()) it = ring_.begin();

        std::vector<ServerNode> result;
        std::unordered_set<std::string> seen;
        auto cur = it;

        while (static_cast<int>(result.size()) < count) {
            if (seen.find(cur->second->id) == seen.end()) {
                seen.insert(cur->second->id);
                result.push_back(*cur->second);
            }
            ++cur;
            if (cur == ring_.end()) cur = ring_.begin();
            if (cur == it) break;  // 環繞一圈
        }
        return result;
    }

    // 分布統計
    std::unordered_map<std::string, int> getDistribution() const {
        std::unordered_map<std::string, int> dist;
        for (const auto& [hash, node] : ring_) {
            dist[node->id]++;
        }
        return dist;
    }

    int nodeCount() const { return static_cast<int>(nodes_.size()); }
    int virtualNodeCount() const { return static_cast<int>(ring_.size()); }

private:
    // FNV-1a Hash（快速且分布均勻）
    uint32_t hashKey(const std::string& key) const {
        uint32_t hash = 2166136261u;  // FNV offset basis
        for (char c : key) {
            hash ^= static_cast<uint8_t>(c);
            hash *= 16777619u;  // FNV prime
        }
        return hash;
    }

    std::map<uint32_t, ServerNode*> ring_;              // 有序環
    std::unordered_map<std::string, ServerNode> nodes_;  // 節點儲存
    int virtualNodesPerWeight_;
};

// ================================================================
// 2. Weighted Round Robin（平滑加權輪詢，執行緒安全）
// ================================================================
class WeightedRoundRobin {
public:
    explicit WeightedRoundRobin(std::vector<ServerNode> servers)
        : servers_(std::move(servers)) {
        totalWeight_ = 0;
        for (const auto& s : servers_) totalWeight_ += s.weight;
        currentWeights_.resize(servers_.size(), 0);
    }

    const ServerNode* next() {
        if (servers_.empty()) return nullptr;

        std::lock_guard<std::mutex> lock(mutex_);

        // 所有節點 current_weight += weight
        for (size_t i = 0; i < servers_.size(); i++) {
            currentWeights_[i] += servers_[i].weight;
        }

        // 選最大的
        int maxIdx = 0;
        for (size_t i = 1; i < servers_.size(); i++) {
            if (currentWeights_[i] > currentWeights_[maxIdx]) {
                maxIdx = static_cast<int>(i);
            }
        }

        // 選中後減去 total_weight
        currentWeights_[maxIdx] -= totalWeight_;
        return &servers_[maxIdx];
    }

private:
    std::vector<ServerNode> servers_;
    std::vector<int> currentWeights_;
    int totalWeight_;
    std::mutex mutex_;
};

// ================================================================
// 3. Least Connections（原子計數器，無鎖高效能）
// ================================================================
class LeastConnections {
public:
    explicit LeastConnections(std::vector<ServerNode> servers)
        : servers_(std::move(servers)) {
        connections_.resize(servers_.size());
        for (auto& conn : connections_) conn.store(0);
    }

    LeastConnections(const LeastConnections&) = delete;  // atomic 不可複製

    const ServerNode* acquire() {
        if (servers_.empty()) return nullptr;

        int minIdx = 0;
        int minConn = connections_[0].load(std::memory_order_relaxed);

        for (size_t i = 1; i < servers_.size(); i++) {
            int conn = connections_[i].load(std::memory_order_relaxed);
            if (conn < minConn) {
                minConn = conn;
                minIdx = static_cast<int>(i);
            }
        }

        connections_[minIdx].fetch_add(1, std::memory_order_relaxed);
        return &servers_[minIdx];
    }

    void release(const std::string& serverId) {
        for (size_t i = 0; i < servers_.size(); i++) {
            if (servers_[i].id == serverId) {
                int prev = connections_[i].fetch_sub(
                    1, std::memory_order_relaxed
                );
                if (prev <= 0) connections_[i].store(0);
                return;
            }
        }
    }

    void printConnections() const {
        for (size_t i = 0; i < servers_.size(); i++) {
            std::cout << servers_[i].id << ": "
                      << connections_[i].load() << " connections\n";
        }
    }

private:
    std::vector<ServerNode> servers_;
    std::vector<std::atomic<int>> connections_;
};

// ================================================================
// 主程式示範
// ================================================================
int main() {
    std::vector<ServerNode> servers = {
        {"server-1", "10.0.0.1", 8080, 2},
        {"server-2", "10.0.0.2", 8080, 4},
        {"server-3", "10.0.0.3", 8080, 2},
    };

    // === Consistent Hash Ring ===
    std::cout << "=== Consistent Hash Ring ===\n";
    ConsistentHashRing ring(150);
    for (const auto& s : servers) ring.addNode(s);

    std::cout << "虛擬節點總數: " << ring.virtualNodeCount() << "\n";
    // 輸出：1200（(2+4+2)×150）

    auto dist = ring.getDistribution();
    for (const auto& [id, count] : dist) {
        std::cout << id << ": " << count << " 虛擬節點\n";
    }
    // 輸出：server-1: 300, server-2: 600, server-3: 300

    // 路由測試
    std::vector<std::string> keys = {
        "user-12345", "session-abc", "cache-key-99"
    };
    for (const auto& key : keys) {
        auto node = ring.getNode(key);
        if (node) std::cout << key << " → " << node->id << "\n";
    }

    // 節點移除（僅影響 ~1/3 流量）
    std::cout << "\n移除 server-2 後：\n";
    ring.removeNode("server-2");
    for (const auto& key : keys) {
        auto node = ring.getNode(key);
        if (node) std::cout << key << " → " << node->id << "\n";
    }

    // === Weighted Round Robin ===
    std::cout << "\n=== Weighted Round Robin ===\n";
    WeightedRoundRobin wrr(servers);
    std::unordered_map<std::string, int> counts;
    for (int i = 0; i < 800; i++) {
        const auto* node = wrr.next();
        if (node) counts[node->id]++;
    }
    for (const auto& [id, count] : counts) {
        std::cout << id << ": " << count << "/800 ("
                  << (count * 100 / 800) << "%)\n";
    }
    // 輸出：server-1: 200 (25%), server-2: 400 (50%), server-3: 200 (25%)

    // === Least Connections ===
    std::cout << "\n=== Least Connections ===\n";
    LeastConnections lc(servers);
    const auto* s1 = lc.acquire();  // server-1
    const auto* s2 = lc.acquire();  // server-2
    const auto* s3 = lc.acquire();  // server-3
    lc.printConnections();
    // 輸出：各 1 connection

    if (s1) lc.release(s1->id);
    std::cout << "釋放 " << s1->id << " 後：\n";
    lc.printConnections();
    // 輸出：server-1: 0, server-2: 1, server-3: 1

    return 0;
}

spec:
  terminationGracePeriodSeconds: 30
  containers:
  - name: app
    lifecycle:
      preStop:
        exec:
          command: ["/bin/sh", "-c", "sleep 5 && kill -SIGTERM 1"]
    # readinessProbe 失敗後 → 從 Service Endpoints 移除
    readinessProbe:
      httpGet:
        path: /health
        port: 8080
      periodSeconds: 5
      failureThreshold: 1