Consistent Hashing Explained: The Complete Distributed Systems Guide

Consistent hashing is a technique that distributes data across a cluster of servers such that when a server is added or removed, only 1/N of the keys need to move — where N is the number of servers. Instead of mapping keys to servers with a modulo operation (which causes nearly all keys to remap when server count changes), consistent hashing places both servers and keys on a circular ring. Each key is owned by the first server clockwise from it on the ring. Add a server, and only the keys between the new server and its predecessor move. Remove a server, and its keys absorb into the next server clockwise. Everything else stays exactly where it is.

This property — minimal disruption on topology changes — is why every major distributed database uses it. Cassandra partitions rows with it. DynamoDB built its storage engine on it. Redis Cluster's 16,384-slot scheme is a specialised variant. If you've ever wondered why these systems can add capacity without cache stampedes or re-sharding downtime, consistent hashing is the answer.

Why Normal Hashing Breaks When Servers Change

The simplest approach to data partitioning is modulo hashing: server = hash(key) % number_of_servers.

With 4 servers:

• Key user:100 → hash("user:100") % 4 = 2 → Server 2
• Key user:101 → hash("user:101") % 4 = 3 → Server 3

This works fine until you add a 5th server. The formula becomes % 5. Almost every key now maps to a different server.

The math is brutal. With N servers and you add one more, the fraction of keys that stay on the same server is approximately N / (N+1). With 4 servers → 5 servers: only 80% stay. With 10 → 11: 91% stay. That sounds okay until you realise you're running a cache. That 9-20% of misses all arrive at the backend simultaneously — a cache stampede that can bring down your database.

Consistent hashing keeps that "keys that move" number at 1/N regardless of total cluster size.

How Consistent Hashing Works

Consistent hashing maps both servers and keys onto the same abstract circle — the hash ring — which wraps from 0 to 2³² - 1 (or any fixed range).

Step 1: Place servers on the ring

Each server is hashed by name or IP to get a position on the ring:

• hash("server-A") → position 83,271,042
• hash("server-B") → position 1,432,891,054
• hash("server-C") → position 2,891,034,217

Step 2: Place keys on the ring

Hash each data key to get its position. Then walk clockwise until you hit a server. That server owns the key.

• hash("user:100") → position 100,000,000 → next clockwise server is B (1,432,891,054) → Server B owns user:100

Step 3: Adding a server

Insert Server D at position 500,000,000. Only keys between Server A (83,271,042) and Server D (500,000,000) need to move from Server B to Server D. All other keys are untouched.

Step 4: Removing a server

Remove Server B. All keys Server B owned (positions 83,271,042 to 1,432,891,054) transfer to Server C (the next clockwise server). No other keys move.

The guarantee: on any topology change (add or remove one server), exactly 1/N of keys relocate.

Consistent Hashing Implementation with Virtual Nodes

The basic ring has a fatal flaw: with just three physical servers, random hash positions can leave one server owning 60% of the ring and another owning 5%. The load is wildly uneven.

Virtual nodes fix this. Each physical server gets placed at multiple positions on the ring — typically 150 to 200 positions per server. These virtual slots are called vnodes. The physical server that wins a key lookup is the one backing the closest vnode.

The math of load distribution: With V vnodes per server and N servers, total ring slots = V × N. Each server's expected load is 1/N. The variance decreases as V increases — at V=150, the standard deviation of load across servers drops to roughly ±5% of the mean. At V=1 (no vnodes), variance can exceed ±40%.

Here's a production-grade implementation with virtual nodes, node removal, and replication:

import hashlib
import bisect
from collections import defaultdict
from typing import Optional
 
class ConsistentHashRing:
    def __init__(self, vnodes: int = 150):
        """
        Args:
            vnodes: Number of virtual nodes per physical server.
                    150-200 is the production standard (Cassandra default: 256).
        """
        self.vnodes = vnodes
        self.ring: dict[int, str] = {}       # position -> physical node name
        self.sorted_keys: list[int] = []      # sorted ring positions
        self.nodes: set[str] = set()          # physical node registry
 
    def add_node(self, node: str) -> None:
        """Add a physical node with `vnodes` virtual positions on the ring."""
        if node in self.nodes:
            return
        self.nodes.add(node)
        for i in range(self.vnodes):
            # Each vnode gets a unique key: "node#vnode_index"
            vnode_key = f"{node}#{i}"
            position = self._hash(vnode_key)
            self.ring[position] = node
            bisect.insort(self.sorted_keys, position)
 
    def remove_node(self, node: str) -> None:
        """
        Remove a physical node and all its virtual positions.
        Keys owned by this node transfer to the next clockwise node automatically
        — callers just re-lookup affected keys after removal.
        """
        if node not in self.nodes:
            return
        self.nodes.discard(node)
        for i in range(self.vnodes):
            vnode_key = f"{node}#{i}"
            position = self._hash(vnode_key)
            if position in self.ring:
                del self.ring[position]
                idx = bisect.bisect_left(self.sorted_keys, position)
                if idx < len(self.sorted_keys) and self.sorted_keys[idx] == position:
                    self.sorted_keys.pop(idx)
 
    def get_node(self, key: str) -> Optional[str]:
        """Return the physical node responsible for `key`."""
        if not self.ring:
            return None
        position = self._hash(key)
        idx = bisect.bisect(self.sorted_keys, position)
        if idx == len(self.sorted_keys):
            idx = 0  # wrap around the ring
        return self.ring[self.sorted_keys[idx]]
 
    def get_nodes_for_replication(self, key: str, replicas: int) -> list[str]:
        """
        Return `replicas` distinct physical nodes for a key.
        Walk clockwise from key position, collecting unique physical nodes.
        This is how Cassandra and DynamoDB implement N-way replication.
        """
        if not self.ring or replicas > len(self.nodes):
            return list(self.nodes)
 
        position = self._hash(key)
        idx = bisect.bisect(self.sorted_keys, position)
        if idx == len(self.sorted_keys):
            idx = 0
 
        seen_nodes: set[str] = set()
        result: list[str] = []
 
        for _ in range(len(self.sorted_keys)):
            node = self.ring[self.sorted_keys[idx]]
            if node not in seen_nodes:
                seen_nodes.add(node)
                result.append(node)
                if len(result) == replicas:
                    break
            idx = (idx + 1) % len(self.sorted_keys)
 
        return result
 
    def _hash(self, key: str) -> int:
        # MD5 is fine for ring positions (not a security use case).
        # Use xxHash or MurmurHash3 in performance-critical paths.
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
 
    def get_load_distribution(self) -> dict[str, float]:
        """
        Return estimated load percentage per physical node.
        Useful for verifying vnode count is sufficient.
        """
        counts: dict[str, int] = defaultdict(int)
        for node in self.ring.values():
            counts[node] += 1
        total = len(self.ring)
        return {node: (count / total) * 100 for node, count in counts.items()}
 
 
# Usage
ring = ConsistentHashRing(vnodes=150)
ring.add_node("cache-1")
ring.add_node("cache-2")
ring.add_node("cache-3")
 
# Single node lookup
node = ring.get_node("user:12345")
print(f"user:12345 → {node}")
 
# Replication across 3 nodes (Cassandra RF=3 style)
replicas = ring.get_nodes_for_replication("user:12345", replicas=3)
print(f"Replicas: {replicas}")
 
# Check load balance
dist = ring.get_load_distribution()
for server, pct in sorted(dist.items()):
    print(f"  {server}: {pct:.1f}%")
# With vnodes=150, 3 servers: each should land near 33.3% ± 2%

Choosing vnode count: Cassandra defaults to 256 vnodes per node. For most systems, 150 is sufficient — it brings load variance to under ±5%. Going below 50 vnodes risks noticeable hotspots. Going above 500 has diminishing returns and increases memory overhead for the ring data structure.

Replication with Consistent Hashing

Production databases almost never store a single copy of data. Consistent hashing extends naturally to replication by walking the ring to find N successive physical nodes for each key.

Cassandra's approach (Replication Factor = 3):

1. Hash the partition key to find its position on the ring
2. The primary replica is the first node clockwise (the coordinator)
3. The second replica is the next distinct physical node clockwise
4. The third replica is the node after that

This is exactly what get_nodes_for_replication implements above. Cassandra calls this the token-aware replication strategy. Each node owns a primary range and acts as a replica for the ranges of its two counter-clockwise neighbors.

DynamoDB's variant: Amazon's Dynamo paper (2007) describes assigning each key to a preference list of N nodes. The first healthy node in the preference list handles writes. During node failures, a technique called sloppy quorum allows a temporarily unavailable node's slot to be filled by the next available node — the written data is later handed back via a process called hinted handoff.

Key "order:9876" → hash position 2,100,000,000
 
Ring positions:
  node-A: 1,800,000,000   ← primary (first clockwise)
  node-B: 2,500,000,000   ← replica 2
  node-C: 3,200,000,000   ← replica 3
  node-D: 500,000,000     ← replica 4 (if RF=4)
 
RF=3: write to node-A, node-B, node-C
      quorum write needs 2 of 3 ACKs

Rack/zone awareness: Production Cassandra uses NetworkTopologyStrategy, which walks the ring but skips vnodes on the same rack until it finds replicas on different racks. This ensures a rack failure doesn't lose all replicas for a key.

Consistent Hashing vs Rendezvous Hashing

Consistent hashing isn't the only algorithm for stable key-to-server mapping. Rendezvous hashing (also called Highest Random Weight hashing) takes a different approach: for each key, compute a score for every server, and assign the key to the server with the highest score.

import hashlib
 
def rendezvous_get_node(key: str, servers: list[str]) -> str:
    """Rendezvous (HRW) hashing — no ring, just score every server."""
    def score(server: str) -> int:
        combined = f"{key}:{server}"
        return int(hashlib.md5(combined.encode()).hexdigest(), 16)
 
    return max(servers, key=score)

When to use consistent hashing:

• Large clusters (50+ nodes) where O(N) lookup is too slow
• Systems where you need heterogeneous node weights via vnode count
• Building a distributed database or cache (the industry standard approach)

When to use rendezvous hashing:

• Small to medium clusters (under 20 nodes) where O(N) is acceptable
• Simpler codebases where you want to avoid ring data structure management
• CDN and load balancer scenarios where you need weighted servers without vnode complexity

Real Production Example: Building a Distributed Cache

Here's how consistent hashing powers a Redis-like distributed cache layer in practice.

Scenario: You have a web application generating 50,000 cache lookups per second. You want to cache across 6 Redis nodes for horizontal scale. Keys include user sessions, product data, and API responses.

import redis
from consistent_hash_ring import ConsistentHashRing  # the class from above
 
class DistributedCache:
    def __init__(self, nodes: list[str], vnodes: int = 150, replicas: int = 1):
        self.ring = ConsistentHashRing(vnodes=vnodes)
        self.connections: dict[str, redis.Redis] = {}
        self.replicas = replicas
 
        for node in nodes:
            host, port = node.split(":")
            self.ring.add_node(node)
            self.connections[node] = redis.Redis(
                host=host,
                port=int(port),
                decode_responses=True,
                socket_connect_timeout=1,
                socket_timeout=1,
            )
 
    def get(self, key: str) -> str | None:
        node = self.ring.get_node(key)
        try:
            return self.connections[node].get(key)
        except redis.RedisError:
            # Node failure: try next node in ring
            return self._get_with_failover(key, failed_node=node)
 
    def set(self, key: str, value: str, ttl: int = 3600) -> bool:
        if self.replicas > 1:
            # Write to multiple nodes for redundancy
            target_nodes = self.ring.get_nodes_for_replication(key, self.replicas)
            results = []
            for node in target_nodes:
                try:
                    results.append(self.connections[node].set(key, value, ex=ttl))
                except redis.RedisError:
                    results.append(False)
            return any(results)  # success if at least one node wrote
        else:
            node = self.ring.get_node(key)
            try:
                return bool(self.connections[node].set(key, value, ex=ttl))
            except redis.RedisError:
                return False
 
    def add_node(self, node: str) -> None:
        """
        Scale out: add a cache node.
        Only ~1/N keys will miss on next lookup (they're now on a different node).
        Those misses will repopulate from the backend — no stampede because
        it's a gradual shift, not a full invalidation.
        """
        host, port = node.split(":")
        self.connections[node] = redis.Redis(host=host, port=int(port))
        self.ring.add_node(node)
 
    def remove_node(self, node: str) -> None:
        """Scale in or handle node failure."""
        self.ring.remove_node(node)
        if node in self.connections:
            self.connections[node].close()
            del self.connections[node]
 
    def _get_with_failover(self, key: str, failed_node: str) -> str | None:
        """Try successive ring nodes on failure."""
        candidates = self.ring.get_nodes_for_replication(key, replicas=3)
        for node in candidates:
            if node == failed_node:
                continue
            try:
                return self.connections[node].get(key)
            except redis.RedisError:
                continue
        return None
 
 
# Deploy: 6 Redis nodes, 150 vnodes each, no replication (single-copy cache)
cache = DistributedCache(
    nodes=["10.0.0.1:6379", "10.0.0.2:6379", "10.0.0.3:6379",
           "10.0.0.4:6379", "10.0.0.5:6379", "10.0.0.6:6379"],
    vnodes=150,
    replicas=1,
)
 
# Adding a 7th node at peak traffic — only ~14% of keys will miss
cache.add_node("10.0.0.7:6379")

What happens at the 14% miss rate? Those keys simply re-fetch from the origin (database, API) and repopulate the new cache node. Because the misses are spread across all keys uniformly (not all at once), backend load increases by roughly 14% temporarily, then returns to baseline as the new node warms up. Compare this to modulo hashing where you'd see a 100% miss spike simultaneously.

Common Pitfalls

1. Too few virtual nodes → hotspots

The most common mistake. With 3 physical servers and no vnodes (V=1), random hash placement can leave one server owning 60% of the ring. Always use at least 100 vnodes per server. Verify with get_load_distribution() before going to production.

2. Wrong hash function

MD5 is fine for ring positions — you don't need cryptographic strength here. The important property is uniform distribution across the output space. Avoid CRC32 (poor distribution), SHA-256 (slower than needed), and especially language built-in hash() functions which are randomised per-process in Python 3 (PYTHONHASHSEED). Use MD5, MurmurHash3, or xxHash. MurmurHash3 is fastest for this use case.

# Don't use this — non-deterministic across processes
position = hash(key) % ring_size  # WRONG: Python hash() is randomised
 
# Use this
import hashlib
position = int(hashlib.md5(key.encode()).hexdigest(), 16)  # OK

3. Not handling node failure gracefully

When a node fails, the ring instantly redirects its keys to the next clockwise node. That node now handles double the traffic. If that node is already near capacity, you've cascaded the failure. Mitigations:

• Monitor per-node load; alert before any node hits 70% capacity
• Use replicas (get_nodes_for_replication) so a failed node's data is available elsewhere
• Implement circuit breakers that skip failed nodes in the ring traversal

4. Hotspot keys that don't distribute

Consistent hashing distributes across servers by key hash. But if one key is accessed 1000x more than others (a "hot key" — say, a celebrity's profile), it still lands on a single node regardless of vnodes. Vnodes don't help here. Solutions:

• Cache the hot key on all nodes with a random suffix strategy: user:celebrity#shard_0, user:celebrity#shard_1, etc., then randomly select a shard on read
• Use a separate dedicated node or local in-memory cache for known hot keys

5. Forgetting that ring state must be consistent across clients

Every client that talks to your distributed cache must have the same view of the ring (same nodes, same vnode count, same hash function). If one client adds a node before others, it sends requests to a node that other clients don't know is responsible for those keys. Use a coordination service (ZooKeeper, etcd, or Consul) to propagate ring membership changes atomically.

Where Consistent Hashing Is Used in Production

Cassandra is the canonical example. Each node owns a set of token ranges on the ring, determined by the partitioner (Murmur3Partitioner by default). The vnodes setting in cassandra.yaml defaults to 256. When you add a node with nodetool bootstrap, Cassandra calculates which existing nodes must stream token ranges to the new node — roughly 1/N of each existing node's data.

DynamoDB uses consistent hashing internally across its storage nodes but abstracts it from users. The Dynamo paper describes preference lists, sloppy quorums, and hinted handoff — all built on top of the ring abstraction.

Redis Cluster uses a variant: 16,384 fixed hash slots. Every key maps to CRC16(key) % 16384. Slots are then assigned to nodes. This is consistent hashing with a fixed vnode count (16,384 total slots) pre-allocated. Adding a node means migrating specific slots, not recomputing positions.

Memcached client libraries (libmemcached, twemproxy) use consistent hashing client-side to route requests to the right server. No server-side coordination needed — the hash ring is embedded in the client.

CDNs like Akamai use consistent hashing to decide which edge node caches a given URL. This ensures the same URL always routes to the same edge node (maximising cache hit rate) until topology changes.

Consistent Hashing vs Mod Hashing: Quick Reference

FAQ

What is consistent hashing and why does it matter?

Consistent hashing is a data distribution technique that maps keys and servers onto a circular ring. When you add or remove a server, only approximately 1/N of the total keys need to move (where N is the number of servers). This matters because the alternative — modulo hashing — remaps nearly all keys on any server change, causing cache stampedes, rebalancing storms, and temporary service degradation. Consistent hashing is what allows systems like Cassandra and DynamoDB to scale horizontally without downtime.

How many virtual nodes should I use per server?

The production standard is 150 to 256 vnodes per physical server. Cassandra defaults to 256. At 150 vnodes, load imbalance across servers stays within ±5% of the mean. Below 50 vnodes you risk significant hotspots; above 500 the memory overhead of the ring grows without meaningful load improvement. When servers have heterogeneous capacity (one has 2x the RAM), give the larger server proportionally more vnodes (2x as many).

What hash function should I use for consistent hashing?

MurmurHash3 is the best choice for performance — it's fast, has excellent uniform distribution, and is not cryptographic (so no unnecessary overhead). MD5 works fine and is available in every standard library. Avoid: Python's built-in hash() (randomised per process), CRC32 (poor distribution), and SHA-256 (cryptographic overhead you don't need). The key requirement is determinism across processes and machines — the same input must always produce the same output.

How does consistent hashing handle node failures?

When a node fails, the ring automatically redirects its keys to the next clockwise node (or nodes, if using replication). If you stored replicas on multiple nodes with get_nodes_for_replication, reads can be served from a replica immediately. The failed node's primary keys now land on the next node, which may temporarily see increased load. The common mitigation is to maintain headroom (never run nodes above 60-70% capacity) and use replication so no single node failure causes data unavailability.

What is the difference between consistent hashing and mod hashing?

Mod hashing computes server = hash(key) % N where N is the number of servers. It's simple and perfectly load-balanced, but adding or removing any server changes N, which changes the assignment of almost every key. Consistent hashing uses a fixed ring where servers occupy positions. Only keys between the added/removed server and its predecessor need to move. The tradeoff: consistent hashing requires a more complex data structure (the ring with sorted positions) but delivers stability under topology changes that mod hashing cannot.

How does Cassandra use consistent hashing?

Cassandra uses a token ring where every row is assigned a token based on Murmur3(partition_key). Each Cassandra node owns a set of token ranges. With vnodes enabled (the default since Cassandra 3.0, set to 256 per node), each node owns 256 small token ranges scattered around the ring rather than one contiguous arc. When you add a new node via nodetool bootstrap, Cassandra identifies which existing nodes own tokens that should transfer to the new node and streams those SSTables. Replication uses the NetworkTopologyStrategy to place replicas on nodes in different racks, walking the ring and skipping nodes on the same rack until it has placed RF replicas per datacenter.

When should I NOT use consistent hashing?

Skip consistent hashing when: (1) your server count is fixed and changes less than once a quarter — modulo hashing or static sharding is simpler; (2) you're using a managed service like ElastiCache, Cloud Spanner, or CockroachDB that handles partitioning transparently; (3) your dataset fits on a single node — premature distribution adds complexity without benefit; (4) you have extreme hot-key problems where one key accounts for >5% of traffic — consistent hashing doesn't help with hot keys, only with server distribution. Also reconsider if your team is not yet running distributed systems in production; the operational overhead (ring management, membership propagation, vnode tuning) requires experience to handle safely.

Key Takeaways

• Normal modulo hashing remaps ~80% of keys when you add a server. Consistent hashing remaps ~10% (1/N).
• The hash ring maps both servers and keys to the same circular space. Each key goes to the first server clockwise from it.
• Virtual nodes (vnodes) fix uneven load distribution. Use 150-256 per server. Cassandra defaults to 256.
• Replication extends the ring walk: collect N distinct physical nodes for each key to store N copies.
• MurmurHash3 is the right hash function. Avoid Python's built-in hash() — it's randomised per process.
• Hot keys (celebrity problem) are not solved by consistent hashing — handle them with key sharding or dedicated tiers.
• Ring state must be consistent across all clients. Use etcd or Consul for membership coordination in production.

Related reading: Load Balancing Strategies · Vertical vs Horizontal Scaling