✍️ 필사 모드: Consistent Hashing 完全ガイド 2025: Virtual Nodes、Jump Hash、Rendezvous、分散システムの中核ビルディングブロック

def get_server(key, num_servers):
    return hash(key) % num_servers

# 4 台に分散
get_server("user:1234", 4)  # server 2
get_server("user:5678", 4)  # server 0
get_server("user:9012", 4)  # server 3

get_server("user:1234", 5)  # server 4 (旧 2 → 新 4)
get_server("user:5678", 5)  # server 3 (旧 0 → 新 3)
get_server("user:9012", 5)  # server 2 (旧 3 → 新 2)

graph LR
    A[Key] --> B{ハッシュ方式}
    B -->|modulo| C[80% の key が移動<br/>災害]
    B -->|Consistent Hash| D[20% の key が移動<br/>正常]

        0 / 2^32
            |
     NodeA  |
    ---+----+----+---
    |       |       |
    |       |       |
    NodeD      NodeB
    |       |       |
    |       |       |
    ---+----+----+---
     NodeC  |
            |

import hashlib
from bisect import bisect_right, insort

class ConsistentHashBasic:
    def __init__(self):
        self.ring = {}       # hash → node
        self.sorted_keys = []

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

    def add_node(self, node: str):
        h = self._hash(node)
        self.ring[h] = node
        insort(self.sorted_keys, h)

    def remove_node(self, node: str):
        h = self._hash(node)
        del self.ring[h]
        self.sorted_keys.remove(h)

    def get_node(self, key: str) -> str:
        if not self.ring:
            return None
        h = self._hash(key)
        # 時計回りで最も近いノードを検索
        idx = bisect_right(self.sorted_keys, h)
        if idx == len(self.sorted_keys):
            idx = 0  # リングを一周して先頭へ
        return self.ring[self.sorted_keys[idx]]

# 使用例
ch = ConsistentHashBasic()
for node in ["server-A", "server-B", "server-C", "server-D"]:
    ch.add_node(node)

print(ch.get_node("user:1234"))  # server-C
print(ch.get_node("user:5678"))  # server-A

NodeA: 0 〜 5 (5%)
NodeB: 5 〜 40 (35%)  ← 多く割り当て
NodeC: 40 〜 50 (10%)
NodeD: 50 〜 100 (50%) ← 半分を占有

物理ノード 4 個(各 150 vnode)

リング上の分布:
A-1, B-7, C-3, A-99, D-42, B-15, C-88, D-1, A-15, ...
(ほぼ均等に混ざる)

class ConsistentHashWithVNodes:
    def __init__(self, vnodes_per_node: int = 150):
        self.vnodes_per_node = vnodes_per_node
        self.ring = {}
        self.sorted_keys = []

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

    def add_node(self, node: str, weight: int = 1):
        # weight が大きいほど vnode が多く作られる
        count = self.vnodes_per_node * weight
        for i in range(count):
            vnode_key = f"{node}#{i}"
            h = self._hash(vnode_key)
            self.ring[h] = node
            insort(self.sorted_keys, h)

    def remove_node(self, node: str, weight: int = 1):
        count = self.vnodes_per_node * weight
        for i in range(count):
            vnode_key = f"{node}#{i}"
            h = self._hash(vnode_key)
            del self.ring[h]
            self.sorted_keys.remove(h)

    def get_node(self, key: str) -> str:
        if not self.ring:
            return None
        h = self._hash(key)
        idx = bisect_right(self.sorted_keys, h)
        if idx == len(self.sorted_keys):
            idx = 0
        return self.ring[self.sorted_keys[idx]]

# 使用例: 容量が異なるサーバー群
ch = ConsistentHashWithVNodes(vnodes_per_node=150)
ch.add_node("server-A", weight=2)  # 性能 2 倍 → 300 vnode
ch.add_node("server-B", weight=1)  # 標準 → 150 vnode
ch.add_node("server-C", weight=1)
ch.add_node("server-D", weight=1)

# これで A は全 key の約 40%、残りは各約 20% を受け取る

import random
import statistics

def benchmark(vnodes_per_node: int, num_nodes: int, num_keys: int):
    ch = ConsistentHashWithVNodes(vnodes_per_node)
    for i in range(num_nodes):
        ch.add_node(f"node-{i}")

    counts = {f"node-{i}": 0 for i in range(num_nodes)}
    for k in range(num_keys):
        key = f"key-{random.random()}"
        node = ch.get_node(key)
        counts[node] += 1

    values = list(counts.values())
    avg = statistics.mean(values)
    stddev = statistics.stdev(values)
    max_val = max(values)
    min_val = min(values)

    print(f"vnodes={vnodes_per_node}: "
          f"min={min_val}, max={max_val}, "
          f"stddev={stddev:.0f} ({stddev/avg*100:.1f}% of avg)")

# 1,000,000 key を 10 ノードへ分散
benchmark(vnodes_per_node=1,   num_nodes=10, num_keys=1_000_000)
benchmark(vnodes_per_node=10,  num_nodes=10, num_keys=1_000_000)
benchmark(vnodes_per_node=100, num_nodes=10, num_keys=1_000_000)
benchmark(vnodes_per_node=500, num_nodes=10, num_keys=1_000_000)

vnodes=1:   min=30000, max=250000, stddev=70000 (70.0% of avg)
vnodes=10:  min=60000, max=180000, stddev=35000 (35.0% of avg)
vnodes=100: min=92000, max=110000, stddev=5800  (5.8% of avg)
vnodes=500: min=97000, max=103000, stddev=2000  (2.0% of avg)

def jump_consistent_hash(key: int, num_buckets: int) -> int:
    """
    Google Jump Consistent Hash
    key: ハッシュ化された整数
    num_buckets: bucket(node)の個数
    """
    b = -1
    j = 0
    while j < num_buckets:
        b = j
        key = (key * 2862933555777941757 + 1) & 0xFFFFFFFFFFFFFFFF
        j = int((b + 1) * (1 << 31) / ((key >> 33) + 1))
    return b

# 使用例
import hashlib

def hash_key(key: str) -> int:
    return int(hashlib.md5(key.encode()).hexdigest()[:16], 16)

num_nodes = 10
key = hash_key("user:1234")
bucket = jump_consistent_hash(key, num_nodes)
print(f"user:1234 → bucket {bucket}")

def rendezvous_hash(key: str, nodes: list) -> str:
    """
    各 key を最大の weight を持つノードに割り当てる
    """
    if not nodes:
        return None

    max_weight = -1
    selected_node = None

    for node in nodes:
        combined = f"{key}:{node}"
        h = int(hashlib.md5(combined.encode()).hexdigest(), 16)
        if h > max_weight:
            max_weight = h
            selected_node = node

    return selected_node

# 使用例
nodes = ["server-A", "server-B", "server-C", "server-D"]
print(rendezvous_hash("user:1234", nodes))  # 例: "server-C"

def rendezvous_hash_topk(key: str, nodes: list, k: int) -> list:
    weights = []
    for node in nodes:
        h = int(hashlib.md5(f"{key}:{node}".encode()).hexdigest(), 16)
        weights.append((h, node))
    weights.sort(reverse=True)
    return [node for _, node in weights[:k]]

# 3 個の replica 用に 3 ノードを選択
replicas = rendezvous_hash_topk("user:1234", nodes, k=3)
print(replicas)  # ["server-C", "server-A", "server-D"]

# cassandra.yaml
num_tokens: 256
partitioner: org.apache.cassandra.dht.Murmur3Partitioner

upstream backend {
    hash $request_uri consistent;
    server backend1.example.com;
    server backend2.example.com;
    server backend3.example.com;
}

# Envoy config
load_balancing_policy:
  policies:
  - typed_extension_config:
      name: envoy.load_balancing_policies.ring_hash
      typed_config:
        "@type": type.googleapis.com/envoy.extensions.load_balancing_policies.ring_hash.v3.RingHash
        minimum_ring_size: 1024

def bad_hash(key):
    return hash(key)  # Python 内蔵 hash() は再起動で変わる!

# 悪い
f"{node}-{i}"   # node1-0, node1-1, ...
f"{node}_{i}"   # node1_0, node1_1, ...(別システム利用)

# 良い: チーム/組織標準を文書化
f"{node}#{i}"   # ドキュメントに明記

# 概念レベルの擬似コード
def build_maglev_table(servers, table_size=65537):
    # table_size は素数(prime)を選ぶ
    permutations = {
        s: generate_permutation(s, table_size) for s in servers
    }
    table = [None] * table_size
    next_idx = {s: 0 for s in servers}

    filled = 0
    while filled < table_size:
        for s in servers:
            while table[permutations[s][next_idx[s]]] is not None:
                next_idx[s] += 1
            table[permutations[s][next_idx[s]]] = s
            next_idx[s] += 1
            filled += 1
            if filled == table_size:
                break
    return table

# Prometheus metrics
consistent_hash_node_keys{node="server-A"} 12050
consistent_hash_node_keys{node="server-B"} 10800
consistent_hash_node_keys{node="server-C"} 11200
consistent_hash_node_keys{node="server-D"} 12500