Consistent Hashing: The Algorithm Behind Scalable Distributed Systems

Add one node to a 12-server cache and you invalidate 92% of it in an instant. A 12-node Redis tier uses hash(key) % N to pick a server. Bump N from 12 to 13 and the formula reassigns 12/13 ≈ 92 percent of keys — every reassignment a cache miss, and the database absorbs the full miss storm at the exact moment you were trying to scale up. The fix is consistent hashing, not "add capacity faster."

This is a hashing problem, not a capacity problem. The algorithm assumes a fixed N. Adding or removing a server invalidates that assumption mathematically — and the consistent hashing algorithm published by Karger et al. in 1997^⁠[1] exists specifically to solve it.

When to Use Consistent Hashing vs. Alternatives

Decision rule: Use consistent hashing if cluster membership changes and the cluster is large enough that rendezvous's O(N) per-key scoring matters. For small fixed clusters, hash % N suffices. For small clusters with frequent scaling, rendezvous hashing may be simpler.

The Problem: Why hash % N Fails at Scale

The simplest distribution: server = hash(key) % N. With a fixed N, keys distribute uniformly and lookups are O(1).

When N changes, nearly everything remaps:

Adding one server (12 → 13): 12/13 = 92% of keys remap
Adding one server (100 → 101): 100/101 = 99% of keys remap

This remapping triggers cascading failures: every remapped key becomes a cache miss, overwhelming your backing store, causing timeouts and user-facing errors.

How Consistent Hashing Works

Both servers and keys are hashed onto a fixed-size ring (0 to 2^32-1) — the structure Amazon's Dynamo paper described as the canonical partitioning mechanism for a high-availability key-value store.^⁠[7] To locate a key's server:

1. Hash the key to a position on the ring (yields an integer from 0 to 2^32-1)
2. Walk clockwise around the ring
3. The first server you encounter owns that key

When a server joins at ring position P, only keys that previously mapped to the next server clockwise need to migrate to P. Every other key stays on its current server. With K total keys and N servers, adding one server remaps approximately K/(N+1) keys — the theoretical minimum. That is N times fewer moves than hash % N, which remaps N/(N+1) of all keys.

graph LR
    subgraph Ring ["Hash Ring (0 to 2³²-1)"]
        direction LR
        S1["Server A<br/>pos: 0x1A3F"]
        S2["Server B<br/>pos: 0x5E02"]
        S3["Server C<br/>pos: 0x9B7D"]
    end

    K1(["key: user:42<br/>hash: 0x3C11"]) -->|clockwise| S2
    K2(["key: session:99<br/>hash: 0x7F20"]) -->|clockwise| S3
    K3(["key: cart:17<br/>hash: 0x0E55"]) -->|clockwise| S1

    S1 -.->|"ring order"| S2
    S2 -.->|"ring order"| S3
    S3 -.->|"wraps to"| S1

Naive implementation (Go):

package chash
 
import (
	"hash/crc32"
	"sort"
	"sync"
)
 
type Ring struct {
	mu     sync.RWMutex
	nodes  map[uint32]string // hash position → node ID
	sorted []uint32          // sorted positions for binary search
}
 
func New() *Ring {
	return &Ring{nodes: make(map[uint32]string)}
}
 
func (r *Ring) GetNode(key string) string {
	r.mu.RLock()
	defer r.mu.RUnlock()
 
	if len(r.sorted) == 0 {
		return ""
	}
 
	h := crc32.ChecksumIEEE([]byte(key))
 
	// Binary search: first node position >= h
	idx := sort.Search(len(r.sorted), func(i int) bool {
		return r.sorted[i] >= h
	})
 
	// Wrap around to ring start if past the end
	if idx == len(r.sorted) {
		idx = 0
	}
 
	return r.nodes[r.sorted[idx]]
}
 
func (r *Ring) AddNode(nodeID string) {
	r.mu.Lock()
	defer r.mu.Unlock()
 
	h := crc32.ChecksumIEEE([]byte(nodeID))
	r.nodes[h] = nodeID
	r.sorted = append(r.sorted, h)
	sort.Slice(r.sorted, func(i, j int) bool { return r.sorted[i] < r.sorted[j] })
}

The flaw: with only 3 physical nodes, the ring has 3 arcs. Hash functions rarely place nodes at equal intervals, so load is uneven: for 3 random positions, the expected largest arc is (1/3)(1 + 1/2 + 1/3) = 11/18 ≈ 61% of the ring and the smallest (1/3)(1/3) = 1/9 ≈ 11%.

Virtual Nodes: Even Load Distribution

Production systems place each physical server at multiple ring positions (Dynamo adopted this multi-token-per-node design for the same reason).^⁠[7] This smooths load distribution and prevents the uneven load the naive single-position implementation produces.

Without virtual nodes, 3 physical nodes produce 3 ring positions and the uneven arcs above. Virtual nodes fix this: each physical node gets many replicas.^⁠[1]

graph LR
    subgraph Before["Naive ring (3 nodes, expected arc sizes)"]
        N1["Node A<br/>~61%"] -.-> N2["Node B<br/>~28%"]
        N2 -.-> N3["Node C<br/>~11%"]
        N3 -.-> N1
    end
    subgraph After["Virtual nodes (3 nodes × 150 vnodes)"]
        V1(("A1..A150"))
        V2(("B1..B150"))
        V3(("C1..C150"))
        V1 --- V2
        V2 --- V3
        V3 --- V1
    end
    Before -->|"add 150 vnodes<br/>per physical node"| After

With 150 virtual nodes per physical node in a 10-node cluster:

- 1,500 total ring positions
- Each physical node occupies ~150 positions
- Keys distribute across all 150 positions uniformly
- Each position handles ~0.07% of keys on average
- Per-node standard deviation ≈ 8% (≈ 1/√150)

Java implementation with vnodes:

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.zip.CRC32;
 
public class ConsistentHashRing {
    private final SortedMap<Long, String> ring = new TreeMap<>();
    private final int vnodes;
 
    public ConsistentHashRing(int vnodeCount) {
        this.vnodes = vnodeCount;
    }
 
    private long hash(String key) {
        CRC32 crc = new CRC32();
        crc.update(key.getBytes(StandardCharsets.UTF_8));
        return crc.getValue();
    }
 
    public void addNode(String nodeId) {
        for (int i = 0; i < vnodes; i++) {
            String vnode = nodeId + ":" + i;
            ring.put(hash(vnode), nodeId);
        }
    }
 
    public void removeNode(String nodeId) {
        for (int i = 0; i < vnodes; i++) {
            ring.remove(hash(nodeId + ":" + i));
        }
    }
 
    public String getNode(String key) {
        if (ring.isEmpty()) return null;
 
        long h = hash(key);
        SortedMap<Long, String> tail = ring.tailMap(h);
        long nodeHash = tail.isEmpty() ? ring.firstKey() : tail.firstKey();
        return ring.get(nodeHash);
    }
}

Each physical node adds vnodes entries to the ring. Removing a node removes all its virtual replicas at once.

Bounded-Load Consistent Hashing

Google's 2016 improvement^⁠[8]: limit each node to K/N * (1 + epsilon) keys. When a key hashes to an overloaded node, try the next node clockwise. This prevents any single node from becoming a hot spot.

In practice: pick epsilon = 0.25, the factor Vimeo deployed (nodes can hold 125% of the theoretical average). For a 100-node cluster with 10M keys, each node normally holds 100K keys; a node at 125K overflows new keys to the next node clockwise. ^⁠[8]

Production Checklist

• Use xxHash or MurmurHash3 — not CRC32 or plain FNV-1a (both cluster short sequential inputs such as node:0, node:1) or cryptographic hashes (overkill)
• Set 150–256 virtual nodes per physical node for cache rings — 150 is a safe baseline; Cassandra defaults to 16 since 4.0 (legacy installs use 256)^⁠[9]⁠[4]. Diminishing returns past 256.
• Guard all ring operations with a read-write lock — writes (add/remove) block reads (get-node)
• Test load distribution — collect ring position samples, measure std deviation; expect ≈ 1/√vnodes (≈ 8% at 150)
• Plan rebalancing windows — adding a node triggers ~K/(N+1) key migrations; batch adds in off-peak hours
• Monitor single-node hot keys — consistent hashing distributes keys, not load; a single key referenced 1000x/sec still hits one node
• Handle node failures gracefully — rebalancing to the next node clockwise (or next N replicas in a replicated system)

Production ring with thread-safe ops + a load-distribution test

A complete Ring type that wraps the algorithm with a single RWMutex (one per ring, not per slot — slot-level locking is overkill for this access pattern):

package consistenthash
 
import (
    "sort"
    "strconv"
    "sync"
 
    "github.com/cespare/xxhash/v2"
)
 
type Ring struct {
    mu        sync.RWMutex
    vnodes    int
    slots     []uint32              // sorted slot positions
    owners    map[uint32]string     // slot -> physical node
}
 
func NewRing(vnodes int) *Ring {
    return &Ring{vnodes: vnodes, owners: make(map[uint32]string)}
}
 
// hash takes the top 32 bits of xxHash64. Plain FNV-1a mixes short, similar
// strings ("a:0", "a:1", ...) so poorly that vnodes cluster on the ring.
func hash(s string) uint32 {
    return uint32(xxhash.Sum64String(s) >> 32)
}
 
func (r *Ring) AddNode(node string) {
    r.mu.Lock(); defer r.mu.Unlock()
    for i := 0; i < r.vnodes; i++ {
        slot := hash(node + ":" + strconv.Itoa(i))
        r.slots = append(r.slots, slot)
        r.owners[slot] = node
    }
    sort.Slice(r.slots, func(i, j int) bool { return r.slots[i] < r.slots[j] })
}
 
func (r *Ring) GetNode(key string) string {
    r.mu.RLock(); defer r.mu.RUnlock()
    if len(r.slots) == 0 { return "" }
    h := hash(key)
    // Binary search: first slot >= h, wrap to slots[0] if none.
    idx := sort.Search(len(r.slots), func(i int) bool { return r.slots[i] >= h })
    if idx == len(r.slots) { idx = 0 }
    return r.owners[r.slots[idx]]
}

Removal: same as add but reverse — drop every (node, vnode) slot. The expensive bit is the sort.Slice on rebuild; for high-churn rings, swap to a sorted-map structure (e.g., github.com/igrmk/treemap):

func (r *Ring) RemoveNode(node string) {
    r.mu.Lock(); defer r.mu.Unlock()
    keep := r.slots[:0]
    for _, s := range r.slots {
        if r.owners[s] != node {
            keep = append(keep, s)
        } else {
            delete(r.owners, s)
        }
    }
    r.slots = keep
    // Already sorted; removal preserves order, no re-sort needed.
}

The load-distribution test every ring deserves — paste it into your test suite as a regression guard. Dropping to 16 randomly placed vnodes (≈ 25% stddev, 1/√16)^⁠[2] or swapping in plain FNV-1a fails it:

func TestRing_LoadDistribution(t *testing.T) {
    r := NewRing(150) // 150 vnodes is the production sweet spot
    nodes := []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j"}
    for _, n := range nodes { r.AddNode(n) }
 
    counts := make(map[string]int)
    const N = 1_000_000
    for i := 0; i < N; i++ {
        counts[r.GetNode(strconv.Itoa(i))]++
    }
 
    expected := float64(N) / float64(len(nodes))
    var sumSqDiff float64
    for _, c := range counts {
        d := float64(c) - expected
        sumSqDiff += d * d
    }
    stdDev := math.Sqrt(sumSqDiff/float64(len(nodes))) / expected
    // A well-mixed hash lands near 1/√150 ≈ 8%. 16% (2×) flags a dropped
    // vnode count or a poorly mixed hash.
    if stdDev > 0.16 {
        t.Errorf("load distribution stddev %.4f > 0.16; check vnode count and hash", stdDev)
    }
}

A rebalance-event handler — emit a structured event when a node joins/leaves so dependent caches can pre-warm or invalidate the migrated keys instead of waiting for cold-cache misses:

type RebalanceEvent struct {
    Type       string    // "node_added" | "node_removed"
    Node       string
    Migrated   int       // estimated keys remapped: totalKeys / node count, the K/(N+1) bound from above
    OccurredAt time.Time
}
 
// AddNodeWithEvent wraps AddNode and emits the event once the ring already
// reflects the new layout. AddNode's own r.mu.Lock is released (via defer)
// before it returns, so no lock is held by the time the send below runs —
// a slow consumer blocking on sink can never stall a concurrent GetNode
// call. Migrated approximates the K/(N+1) bound from the intro; the Ring
// type never sees the actual dataset, only slot hashes, so an exact count
// would require the caller to track real keys.
func (r *Ring) AddNodeWithEvent(node string, totalKeys int, sink chan<- RebalanceEvent) {
    r.AddNode(node)
 
    r.mu.RLock()
    nodeCount := len(r.slots) / r.vnodes
    r.mu.RUnlock()
 
    migrated := 0
    if nodeCount > 0 {
        migrated = totalKeys / nodeCount
    }
    sink <- RebalanceEvent{Type: "node_added", Node: node, Migrated: migrated, OccurredAt: time.Now()}
}

Subscribers do their own bookkeeping — the ring stays a pure data structure. The pattern beats baking notification into the ring directly because every consumer wants different semantics (warm-on-add, invalidate-on-remove, log-only).

A ring config in the spirit of Cassandra's num_tokens — vnode count is the knob that decides balance (Cassandra's own 256 → 16 change is covered below):

# ring.yaml — service-side ring configuration
ring:
  vnodes_per_node: 150          # 150 = balanced cache ring; 16 = Cassandra-style
  hash_function: murmur3        # CRC32 has clustering bug for sequential keys
  rebalance_batch_size: 100     # max concurrent migrations during topology change
  rebalance_off_peak_only: true # skip rebalance during business-hours peak

Jump Consistent Hash: When the Ring Is Overkill

Lamping and Veach published Jump Consistent Hash at Google in 2014. The algorithm is about five lines of code, needs no memory beyond a few registers, and in the paper's C++ benchmarks ran 1.6–4.4× faster than a sorted-array ring with 100 points per bucket. The catch: it only supports buckets numbered 0 to N-1. Removing bucket 7 from a 100-bucket ring is impossible without renumbering — the algorithm assumes contiguous bucket IDs. ^⁠[2]

The math is a probabilistic walk: for each bucket count from 1 to N, decide whether the key "jumps" to the new bucket based on a deterministic random sequence seeded by the key. The expected number of iterations is below ln(N) + 1, so fewer than 8 for a 1000-bucket cluster. ^⁠[2]

// JumpHash returns a bucket in [0, numBuckets) for the given key.
// Source: Lamping & Veach, "A Fast, Minimal Memory, Consistent Hash Algorithm" (2014)
func JumpHash(key uint64, numBuckets int32) int32 {
    var b, j int64 = -1, 0
    for j < int64(numBuckets) {
        b = j
        key = key*2862933555777941757 + 1
        j = int64(float64(b+1) * (float64(int64(1)<<31) / float64((key>>33)+1)))
    }
    return int32(b)
}

Use Jump when buckets are append-only (shard IDs in a sharded database, partition IDs in Kafka with manual rebalancing) and you need sub-microsecond lookups. Avoid Jump when nodes can fail and need replacement at arbitrary positions — the renumbering cost defeats the algorithm's elegance.

Real-World Tuning: Memcached, DynamoDB, Cassandra

The three production rings every backend engineer eventually touches differ more than the textbook suggests. Their tuning choices reveal what each system actually optimizes for.

Memcached (libketama): 160 vnodes per node, MD5-derived 32-bit positions. The libketama algorithm used by pylibmc (via libmemcached) and spymemcached predates the consistent-hashing-with-bounded-loads era — it has no bounded-load enforcement. Hot keys land where they land. Production deployments add a level-1 in-process LRU on each app server to absorb hot-key traffic before the ring lookup ever runs. Twemproxy implements libketama-compatible distribution so ring decisions stay consistent across client libraries; mcrouter does not (it defaults to its own ch3 hash), so don't mix it with ketama clients on one pool.^⁠[10]

DynamoDB: the managed service doesn't document its ring. The original Dynamo paper hashed keys with MD5 into a 128-bit space, gave each node multiple virtual nodes, stored each key on the first N nodes clockwise from its position ((N, R, W) = (3, 2, 2) was the common setting), and found fixed equal-sized partitions balanced load best.^⁠[7] Today's DynamoDB hides all of it: a partition serves at most 3,000 RCUs and 1,000 WCUs, and adaptive capacity boosts hot partitions and rebalances them so frequently accessed items don't share one.^⁠[11] The lesson: AWS bought you out of vnode tuning by making the ring opaque.

Cassandra: dropped from 256 to 16 vnodes per node in 4.0.^⁠[4] The trade-off is availability, not bootstrap cost: more tokens give a bootstrapping node more streaming peers, but each node then shares data with more peers, which lowers availability. The docs recommend 16 for elastic clusters that grow and shrink regularly, not for clusters over 50 nodes, and stress setting allocate_tokens_for_local_replication_factor so tokens are allocated evenly instead of randomly.^⁠[12]

Bounded-Load vs Power-of-Two-Choices

Both algorithms tackle the same problem — uneven load with naive consistent hashing — with different trade-offs. The choice between them depends on whether you need a hard cap on per-node load or a fixed per-request cost.

Bounded-load (Mirrokni et al., Google 2016): cap each node at K/N * (1 + epsilon) keys; overflow walks clockwise to the next node with room. Per-request cost is one ring lookup plus that walk, and the paper bounds the expected number of nodes visited at O(1/ε²) for ε ≤ 1 — a tighter cap means longer walks. ^⁠[8]

Power-of-two-choices (Mitzenmacher 1996): hash the key with two independent hash functions, pick the less-loaded of the two candidate nodes. Cost is always exactly two lookups. With one choice and K = N, the busiest node holds Θ(log N / log log N) keys; with two choices the excess over the average drops to O(log log N) with high probability. ^⁠[8]

What each scheme guarantees, per the bounded-loads paper's analysis:

Bounded-load is the only one with a hard cap, which is why Vimeo moved to it (c = 1.25) after giving up on plain consistent hashing and power-of-two-choices for its video-streaming load balancing. ^⁠[8] Power-of-two-choices keeps a fixed two-lookup cost but bounds the maximum only with high probability.

A minimal power-of-two-choices selector on top of a vnode ring. Suffixing the key with :a and :b gives two independent ring positions, so the candidates coincide only about 1/N of the time:

type LoadAwareRing struct {
    *Ring
    mu   sync.Mutex
    load map[string]*atomic.Int64 // node -> in-flight key count
}
 
// AddNode wraps Ring.AddNode and seeds a load counter for the node before
// it becomes reachable through GetNode. PickNode indexes r.load by node
// name; a lookup miss on a map of *atomic.Int64 returns a nil pointer, and
// calling Load() on it panics.
func (r *LoadAwareRing) AddNode(node string) {
    r.mu.Lock()
    if r.load == nil {
        r.load = make(map[string]*atomic.Int64)
    }
    if _, ok := r.load[node]; !ok {
        r.load[node] = &atomic.Int64{}
    }
    r.mu.Unlock()
    r.Ring.AddNode(node)
}
 
func (r *LoadAwareRing) PickNode(key string) string {
    a := r.GetNode(key + ":a")
    b := r.GetNode(key + ":b")
    r.mu.Lock()
    aLoad, bLoad := r.load[a], r.load[b]
    r.mu.Unlock()
    switch {
    case aLoad == nil: // empty ring (a == ""), or a node added via a promoted *Ring method
        return b
    case bLoad == nil:
        return a
    case aLoad.Load() <= bLoad.Load():
        return a
    }
    return b
}

AddNode seeds a counter before a node becomes reachable through GetNode, but PickNode still nil-checks: an empty ring returns "", and a node that joins through a promoted *Ring method such as AddNodeWithEvent skips this override — Go embedding has no virtual dispatch. The short mu lock guards only the map lookup; the two Load() calls race with concurrent increments, so the worst case is picking the slightly-worse node, never a panic.

Frequently Asked Questions

Keep Reading

• Rate Limiter Algorithms — distributed token bucket and sliding window patterns
• Caching Strategies at Scale — consistent hashing in caches and CDNs
• Building Resilient Distributed Systems with Go — partitioning and replication patterns
• Scaling Redis for High-Throughput Systems — Redis Cluster's 16,384 hash slots are a discrete consistent-hashing variant; same hot-key shape, same rebalancing math
• Vector Databases Comparison — sharded vector indexes (Milvus, Pinecone) use the same hash-ring partitioning to balance ANN load across nodes