⚙️ 核心组件详解¶

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

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

class WeightedRoundRobinBalancer:
    def __init__(self, servers_with_weights):
        """servers_with_weights: [('A', 5), ('B', 3), ('C', 2)]"""
        self.servers = servers_with_weights
        self.current_weights = [0] * len(servers_with_weights)
        self.total_weight = sum(w for _, w in servers_with_weights)

    def get_server(self):
        # Nginx smooth weighted round robin
        max_weight = -1
        max_index = 0
        for i, (server, weight) in enumerate(self.servers):  # enumerate同时获取索引和值
            self.current_weights[i] += weight
            if self.current_weights[i] > max_weight:
                max_weight = self.current_weights[i]
                max_index = i

        self.current_weights[max_index] -= self.total_weight
        return self.servers[max_index][0]

import heapq
from collections import defaultdict  # defaultdict带默认值的字典，避免KeyError

class LeastConnectionsBalancer:
    def __init__(self, servers):
        self.connections = {s: 0 for s in servers}

    def get_server(self):
        server = min(self.connections, key=self.connections.get)
        self.connections[server] += 1
        return server

    def release_server(self, server):
        self.connections[server] -= 1

import hashlib
import bisect

class ConsistentHash:
    def __init__(self, nodes=None, replicas=150):
        """replicas: 每个真实节点的虚拟节点数"""
        self.replicas = replicas
        self.ring = {}       # hash_value -> node
        self.sorted_keys = []  # 排序的hash值列表

        if nodes:
            for node in nodes:
                self.add_node(node)

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

    def add_node(self, node):
        for i in range(self.replicas):
            virtual_key = f"{node}:v{i}"
            hash_val = self._hash(virtual_key)
            self.ring[hash_val] = node
            bisect.insort(self.sorted_keys, hash_val)

    def remove_node(self, node):
        for i in range(self.replicas):
            virtual_key = f"{node}:v{i}"
            hash_val = self._hash(virtual_key)
            del self.ring[hash_val]
            self.sorted_keys.remove(hash_val)

    def get_node(self, key):
        if not self.ring:
            return None
        hash_val = self._hash(key)
        idx = bisect.bisect_right(self.sorted_keys, hash_val)
        if idx == len(self.sorted_keys):
            idx = 0
        return self.ring[self.sorted_keys[idx]]

class CacheAsideService:
    def __init__(self, cache, db):
        self.cache = cache
        self.db = db

    def get(self, key):
        # 1. 查缓存
        value = self.cache.get(key)
        if value is not None:
            return value  # 命中

        # 2. 未命中，查数据库
        value = self.db.get(key)
        if value is not None:
            # 3. 写入缓存
            self.cache.set(key, value, ttl=300)
        return value

    def update(self, key, value):
        # 1. 更新数据库
        self.db.update(key, value)
        # 2. 删除缓存（而不是更新缓存！）
        self.cache.delete(key)

from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.cache = OrderedDict()

    def get(self, key):
        if key not in self.cache:
            return None
        self.cache.move_to_end(key)  # 移到末尾（最近使用）
        return self.cache[key]

    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)  # 移除最久未使用

import time
import threading

class SnowflakeGenerator:
    # 起始时间戳 (2020-01-01 00:00:00 UTC)
    EPOCH = 1577836800000

    # 各部分位数
    WORKER_ID_BITS = 5
    DATACENTER_ID_BITS = 5
    SEQUENCE_BITS = 12

    # 最大值
    MAX_WORKER_ID = (1 << WORKER_ID_BITS) - 1      # 31
    MAX_DATACENTER_ID = (1 << DATACENTER_ID_BITS) - 1  # 31
    MAX_SEQUENCE = (1 << SEQUENCE_BITS) - 1          # 4095

    # 位移量
    WORKER_ID_SHIFT = SEQUENCE_BITS                  # 12
    DATACENTER_ID_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS  # 17
    TIMESTAMP_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS + DATACENTER_ID_BITS  # 22

    def __init__(self, datacenter_id, worker_id):
        if datacenter_id > self.MAX_DATACENTER_ID or datacenter_id < 0:
            raise ValueError(f"Datacenter ID must be between 0 and {self.MAX_DATACENTER_ID}")
        if worker_id > self.MAX_WORKER_ID or worker_id < 0:
            raise ValueError(f"Worker ID must be between 0 and {self.MAX_WORKER_ID}")

        self.datacenter_id = datacenter_id
        self.worker_id = worker_id
        self.sequence = 0
        self.last_timestamp = -1
        self.lock = threading.Lock()

    def _current_millis(self):
        return int(time.time() * 1000)

    def next_id(self):
        with self.lock:
            timestamp = self._current_millis()

            if timestamp == self.last_timestamp:
                self.sequence = (self.sequence + 1) & self.MAX_SEQUENCE
                if self.sequence == 0:
                    # 序列号用完，等待下一毫秒
                    while timestamp <= self.last_timestamp:
                        timestamp = self._current_millis()
            else:
                self.sequence = 0

            if timestamp < self.last_timestamp:
                raise Exception("Clock moved backwards!")

            self.last_timestamp = timestamp

            return (
                ((timestamp - self.EPOCH) << self.TIMESTAMP_SHIFT) |
                (self.datacenter_id << self.DATACENTER_ID_SHIFT) |
                (self.worker_id << self.WORKER_ID_SHIFT) |
                self.sequence
            )

# 使用示例
generator = SnowflakeGenerator(datacenter_id=1, worker_id=1)
id1 = generator.next_id()
id2 = generator.next_id()

import time
import threading

class TokenBucket:
    def __init__(self, rate, capacity):
        """
        rate: 每秒生成令牌速率
        capacity: 桶的最大容量
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity  # 初始满桶
        self.last_time = time.time()
        self.lock = threading.Lock()

    def allow_request(self):
        with self.lock:
            now = time.time()
            # 补充令牌
            elapsed = now - self.last_time
            self.tokens = min(
                self.capacity,
                self.tokens + elapsed * self.rate
            )
            self.last_time = now

            # 尝试消耗一个令牌
            if self.tokens >= 1:
                self.tokens -= 1
                return True
            return False

# 使用：每秒10个请求，允许突发20个
limiter = TokenBucket(rate=10, capacity=20)

# 模拟请求
for i in range(25):
    if limiter.allow_request():
        print(f"Request {i}: Allowed")
    else:
        print(f"Request {i}: Rejected")

import time
import threading
from collections import deque

class LeakyBucket:
    def __init__(self, rate, capacity):
        """
        rate: 每秒处理请求数（漏出速率）
        capacity: 桶的最大容量
        """
        self.rate = rate
        self.capacity = capacity
        self.water = 0  # 当前水量（排队请求数）
        self.last_time = time.time()
        self.lock = threading.Lock()

    def allow_request(self):
        with self.lock:
            now = time.time()
            elapsed = now - self.last_time

            # 漏掉一些水
            self.water = max(0, self.water - elapsed * self.rate)
            self.last_time = now

            # 尝试加水
            if self.water < self.capacity:
                self.water += 1
                return True
            return False

import time
import threading  # 线程池/多线程：并发执行任务
from collections import deque

class SlidingWindowLog:
    """滑动窗口日志算法 - 精确但内存开销大"""
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = deque()  # 存储每个请求的时间戳
        self.lock = threading.Lock()

    def allow_request(self):
        with self.lock:
            now = time.time()
            # 清除窗口外的请求
            while self.requests and self.requests[0] < now - self.window_seconds:
                self.requests.popleft()

            if len(self.requests) < self.max_requests:
                self.requests.append(now)
                return True
            return False

class SlidingWindowCounter:
    """滑动窗口计数器 - 近似但内存省"""
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.current_count = 0
        self.previous_count = 0
        self.current_start = time.time()
        self.lock = threading.Lock()

    def allow_request(self):
        with self.lock:
            now = time.time()
            elapsed = now - self.current_start

            if elapsed >= self.window_seconds:
                self.previous_count = self.current_count
                self.current_count = 0
                self.current_start = now
                elapsed = 0

            # 加权计算
            weight = 1 - (elapsed / self.window_seconds)
            estimated_count = self.previous_count * weight + self.current_count

            if estimated_count < self.max_requests:
                self.current_count += 1
                return True
            return False