【网络工程】负载均衡算法深度解析：从轮询到 P2C

class RoundRobin:
    def __init__(self, servers):
        self.servers = servers
        self.index = 0

    def next(self):
        server = self.servers[self.index % len(self.servers)]
        self.index += 1
        return server

# 使用
rr = RoundRobin(["A", "B", "C"])
# 请求序列: A, B, C, A, B, C, A, ...

class WeightedRoundRobin:
    """Nginx 的平滑加权轮询算法"""
    def __init__(self, servers):
        # servers: [(name, weight), ...]
        self.servers = servers
        self.weights = [w for _, w in servers]
        self.current_weights = [0] * len(servers)
        self.total_weight = sum(self.weights)

    def next(self):
        # 1. 每个节点的 current_weight 加上自身 weight
        for i in range(len(self.servers)):
            self.current_weights[i] += self.weights[i]

        # 2. 选择 current_weight 最大的节点
        max_idx = self.current_weights.index(max(self.current_weights))

        # 3. 被选中节点的 current_weight 减去 total_weight
        self.current_weights[max_idx] -= self.total_weight

        return self.servers[max_idx][0]

# 权重 A:5, B:1, C:1 的调度序列
wrr = WeightedRoundRobin([("A", 5), ("B", 1), ("C", 1)])
# 序列: A, A, B, A, C, A, A  (7 个请求中 A 得到 5 个)
# Nginx 的算法保证分布均匀，不会出现 A,A,A,A,A,B,C 这种连续分配

import heapq
from threading import Lock

class LeastConnection:
    def __init__(self, servers):
        # (active_connections, server_name)
        self.heap = [(0, s) for s in servers]
        heapq.heapify(self.heap)
        self.lock = Lock()
        self.conn_count = {s: 0 for s in servers}

    def acquire(self):
        with self.lock:
            _, server = heapq.heappop(self.heap)
            self.conn_count[server] += 1
            heapq.heappush(self.heap, (self.conn_count[server], server))
            return server

    def release(self, server):
        with self.lock:
            self.conn_count[server] -= 1
            # 重建堆（简化实现，实际用索引堆更高效）
            self.heap = [(self.conn_count[s], s) for s in self.conn_count]
            heapq.heapify(self.heap)

def ip_hash(client_ip, servers):
    h = hash(client_ip) % len(servers)
    return servers[h]

import hashlib
import bisect

class ConsistentHash:
    def __init__(self, servers, replicas=150):
        self.ring = []            # 排序的哈希值列表
        self.hash_to_server = {}  # 哈希值 → 服务器映射
        self.replicas = replicas  # 每个服务器的虚拟节点数

        for server in servers:
            self.add_server(server)

    def _hash(self, key):
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def add_server(self, server):
        for i in range(self.replicas):
            h = self._hash(f"{server}:{i}")
            self.ring.append(h)
            self.hash_to_server[h] = server
        self.ring.sort()

    def remove_server(self, server):
        for i in range(self.replicas):
            h = self._hash(f"{server}:{i}")
            self.ring.remove(h)
            del self.hash_to_server[h]

    def get_server(self, key):
        if not self.ring:
            return None
        h = self._hash(key)
        idx = bisect.bisect_right(self.ring, h) % len(self.ring)
        return self.hash_to_server[self.ring[idx]]

# 当移除 Server B 时：
# 只有原本映射到 B 的键会重新映射到 B 的后继节点
# 其他键的映射不变
# 理论上只有 1/N 的键需要重新映射

class BoundedLoadConsistentHash(ConsistentHash):
    """有界负载一致性哈希"""
    def __init__(self, servers, replicas=150, load_factor=1.25):
        super().__init__(servers, replicas)
        self.load_factor = load_factor
        self.load = {s: 0 for s in servers}
        self.total_load = 0

    def _max_load(self):
        """每台服务器的负载上限"""
        avg = self.total_load / len(self.load) if self.load else 0
        return max(1, int(avg * self.load_factor))

    def get_server(self, key):
        h = self._hash(key)
        idx = bisect.bisect_right(self.ring, h) % len(self.ring)
        max_load = self._max_load()

        # 沿环顺时针查找，跳过已超载的节点
        for _ in range(len(self.ring)):
            server = self.hash_to_server[self.ring[idx]]
            if self.load.get(server, 0) < max_load:
                self.load[server] = self.load.get(server, 0) + 1
                self.total_load += 1
                return server
            idx = (idx + 1) % len(self.ring)

        # fallback: 选负载最小的
        return min(self.load, key=self.load.get)

import random

class P2C:
    """Power of Two Choices 负载均衡"""
    def __init__(self, servers):
        self.servers = {s: 0 for s in servers}  # server → active requests

    def pick(self):
        servers = list(self.servers.keys())

        if len(servers) == 1:
            return servers[0]

        # 随机选两个不同的服务器
        a, b = random.sample(servers, 2)

        # 返回负载较低的
        if self.servers[a] <= self.servers[b]:
            self.servers[a] += 1
            return a
        else:
            self.servers[b] += 1
            return b

    def done(self, server):
        self.servers[server] = max(0, self.servers[server] - 1)

// gRPC pick_first → round_robin → P2C (with EWMA latency)
// P2C 结合 EWMA 延迟的伪代码

type Server struct {
    Addr        string
    ActiveReqs  int64
    AvgLatency  float64  // EWMA 延迟
}

func p2cPick(servers []*Server) *Server {
    // 随机选两个
    a := servers[rand.Intn(len(servers))]
    b := servers[rand.Intn(len(servers))]
    for b == a {
        b = servers[rand.Intn(len(servers))]
    }

    // 综合考虑活跃请求数和延迟
    scoreA := float64(a.ActiveReqs+1) * a.AvgLatency
    scoreB := float64(b.ActiveReqs+1) * b.AvgLatency

    if scoreA <= scoreB {
        return a
    }
    return b
}

class P2CRobust:
    """工程化的 P2C 实现"""
    def __init__(self, servers):
        self.servers = {s: {"active": 0, "latency": EWMA()} for s in servers}
        self.last_picked = None

    def pick(self):
        healthy = [s for s in self.servers if self.is_healthy(s)]
        if not healthy:
            return None
        if len(healthy) == 1:
            return healthy[0]

        a, b = random.sample(healthy, 2)

        # 避免连续选同一个（减少突发负载）
        if a == self.last_picked and self.score(a) == self.score(b):
            a, b = b, a

        chosen = a if self.score(a) <= self.score(b) else b
        self.servers[chosen]["active"] += 1
        self.last_picked = chosen
        return chosen

    def score(self, server):
        s = self.servers[server]
        latency = max(s["latency"].get(), 0.001)  # 防止除零
        return (s["active"] + 1) * latency

    def is_healthy(self, server):
        # 健康检查逻辑
        return True

class MaglevHash:
    """简化的 Maglev 一致性哈希"""
    def __init__(self, backends, table_size=65537):
        self.backends = backends
        self.table_size = table_size
        self.table = self._generate_table()

    def _generate_table(self):
        n = len(self.backends)
        # 每个后端生成偏移量和步长
        permutations = []
        for backend in self.backends:
            offset = hash(backend + "_offset") % self.table_size
            skip = hash(backend + "_skip") % (self.table_size - 1) + 1
            permutations.append((offset, skip))

        table = [-1] * self.table_size
        filled = 0
        idx = [0] * n  # 每个后端的当前尝试位置

        while filled < self.table_size:
            for i in range(n):
                offset, skip = permutations[i]
                # 找到下一个空位
                pos = (offset + idx[i] * skip) % self.table_size
                while table[pos] != -1:
                    idx[i] += 1
                    pos = (offset + idx[i] * skip) % self.table_size
                table[pos] = i
                idx[i] += 1
                filled += 1
                if filled >= self.table_size:
                    break

        return table

    def lookup(self, key):
        idx = hash(key) % self.table_size
        return self.backends[self.table[idx]]

class EWMA:
    """指数加权移动平均"""
    def __init__(self, decay=0.95):
        self.value = 0.0
        self.decay = decay
        self.initialized = False

    def update(self, sample):
        if not self.initialized:
            self.value = sample
            self.initialized = True
        else:
            self.value = self.decay * self.value + (1 - self.decay) * sample

    def get(self):
        return self.value

class LatencyAwareLB:
    """延迟感知的 P2C 负载均衡"""
    def __init__(self, servers):
        self.servers = {}
        for s in servers:
            self.servers[s] = {
                "active": 0,
                "latency": EWMA(decay=0.9),
            }

    def pick(self):
        keys = list(self.servers.keys())
        a, b = random.sample(keys, 2)

        sa = self.servers[a]
        sb = self.servers[b]

        # score = 活跃请求数 × 平均延迟
        score_a = (sa["active"] + 1) * max(sa["latency"].get(), 0.001)
        score_b = (sb["active"] + 1) * max(sb["latency"].get(), 0.001)

        chosen = a if score_a <= score_b else b
        self.servers[chosen]["active"] += 1
        return chosen

    def done(self, server, latency_ms):
        self.servers[server]["active"] -= 1
        self.servers[server]["latency"].update(latency_ms)

# HAProxy — 算法配置

backend api_rr
    balance roundrobin
    server s1 10.0.2.10:8080 check weight 1
    server s2 10.0.2.11:8080 check weight 1

backend api_leastconn
    balance leastconn
    server s1 10.0.2.10:8080 check weight 3
    server s2 10.0.2.11:8080 check weight 1

backend cache_consistent
    balance uri                    # URI 哈希
    hash-type consistent
    server s1 10.0.2.10:8080 check
    server s2 10.0.2.11:8080 check

backend api_random_p2c
    balance random(2)              # HAProxy 2.4+ 支持 P2C
    server s1 10.0.2.10:8080 check
    server s2 10.0.2.11:8080 check

# Envoy — P2C + EWMA (默认)
clusters:
- name: api_cluster
  lb_policy: LEAST_REQUEST        # Envoy 的 LEAST_REQUEST 本质是 P2C
  least_request_lb_config:
    choice_count: 2               # 默认 2，即 P2C
  load_assignment:
    cluster_name: api_cluster
    endpoints:
    - lb_endpoints:
      - endpoint:
          address:
            socket_address:
              address: 10.0.2.10
              port_value: 8080
      - endpoint:
          address:
            socket_address:
              address: 10.0.2.11
              port_value: 8080

# 1. 观察后端负载分布
# Nginx access log 统计各 upstream 的请求分布
awk '{print $NF}' /var/log/nginx/access.log | sort | uniq -c | sort -rn
# 如果某台服务器的请求数远超其他，说明算法选择或权重有问题

# 2. 监控后端响应时间差异
# 通过 Nginx $upstream_response_time 记录
log_format upstream_log '$remote_addr - [$time_local] '
    '"$request" $status $upstream_addr $upstream_response_time';

# 分析各后端的 P50/P99
awk '{split($NF, a, ","); for(i in a) print $6, a[i]}' /var/log/nginx/upstream.log \
    | sort | head -20

# 3. 检查连接复用率
# 如果 TIME_WAIT 很多，说明连接没有被复用
ss -tn state time-wait | awk '{print $4}' | cut -d: -f1 | sort | uniq -c | sort -rn

# 4. 基准测试
# 用 wrk 对比不同算法的吞吐量
wrk -t4 -c100 -d30s http://10.0.1.100/api/test
# 切换算法后重新测试，对比 RPS 和延迟分布