Design a Distributed Cache like Redis / Memcached

Almost every large system you will design (feeds, product catalogues, sessions, rate limiters, leaderboards) puts a cache between services and the database. So "design a distributed cache" tests whether you understand the component you reach for in every other answer. The interviewer is not looking for a feature list of Redis. They want you to reason about partitioning, what happens when nodes join and die, and the ugly concurrency bugs that appear when a cache and a database disagree.

I will design a Memcached/Redis-style cache tier, then go deep on the six topics that decide the interview: consistent hashing, replication, eviction, write policies, stampedes and hot keys.

Requirements

Functional

• get(key), set(key, value, ttl), delete(key); optionally atomic incr and compare-and-set.
• Values up to ~1 MB (larger objects belong in object storage); keys up to 250 bytes.
• Per-key TTL and automatic eviction when memory is full.

Non-functional

• Latency: p99 under 1–2 ms within a data centre.
• Throughput: millions of operations per second across the cluster, scaling horizontally.
• Availability over consistency: a cache miss is acceptable, an unavailable cache tier that takes the database down with it is not.
• Not the source of truth: data loss is tolerable (it can be recomputed), but should be rare to avoid miss storms.

Capacity Estimation

Assume a large e-commerce or social backend (interview numbers):

• Peak read traffic: 2M cache operations/sec; 90% reads.
• Working set: 2 billion items × 1 KB average (key + value + ~60 bytes metadata) ≈ 2 TB.
• Usable memory per node: 64 GB of a 96 GB machine (headroom for fragmentation, replication buffers, OS).

Nodes for memory: 2 TB ÷ 64 GB ≈ 32 primaries. With one replica each, 64 nodes. Plan for 70% fill to absorb growth and rebalancing: ~45 primaries, ~90 nodes total.

Nodes for throughput: a single Redis shard serves on the order of 100K–200K simple ops/sec (more with pipelining and I/O threads); multi-threaded Memcached handles more per box. 2M ops/s ÷ 45 primaries ≈ 45K ops/s per shard, comfortably within limits even before reads go to replicas. Here memory, not CPU, determines cluster size, which is typical.

Network: 2M ops × 1 KB ≈ 2 GB/s ≈ 16 Gbps aggregate, ~360 Mbps per node. Fine on 10 Gbps NICs, but a single hot key serving 100K reads/s of a 100 KB value would be 10 GB/s from one node, which saturates it. Hot keys are a bandwidth problem as much as a CPU problem.

API and Single-Node Design

Inside one node, a cache is a hash table plus an eviction structure. The classic LRU design is a hash map pointing into a doubly linked list: get moves the node to the head, set inserts at the head, eviction pops the tail, all O(1). If you have not coded it, do LeetCode 146: LRU Cache; interviewers sometimes ask for it inside this design.

Key node-level decisions:

• Threading: Memcached uses multiple worker threads with fine-grained locks. Redis executes commands on a single thread (with optional I/O threads), which avoids locks and makes every command atomic, at the cost of using one core per shard (so you run several shards per machine).
• Memory allocation: Memcached's slab allocator groups items by size class to avoid fragmentation; Redis relies on jemalloc and active defragmentation.
• Expiry: lazy (check TTL on access) plus active sampling (periodically test random keys with TTLs and delete expired ones), so memory is reclaimed without scanning everything.
• Protocol: a compact text or binary protocol over persistent TCP connections, with pipelining and multi-get to amortise round trips.

Partitioning with Consistent Hashing

With N nodes, the obvious mapping is node = hash(key) % N. It breaks badly on scaling: going from 10 to 11 nodes changes the target for about 10/11 ≈ 91% of keys. Overnight your hit ratio collapses and every miss lands on the database. That is a self-inflicted outage.

The hash ring

Hash both nodes and keys onto a ring of 2³² (or 2⁶⁴) positions. A key belongs to the first node clockwise from its position. Adding a node only steals the keys between it and its predecessor: on average 1/N of keys move. Removing a node hands its arc to the next node.

Virtual nodes fix uneven arcs. Each physical node appears at 100–200 ring positions ("cache-7#0", "cache-7#1"…). Load evens out statistically, a dead node's keys spread over many survivors instead of one neighbour, and bigger machines can take more vnodes for proportional weight.

import bisect
import hashlib

class ConsistentHashRing:
    def __init__(self, nodes=(), vnodes=150):
        self.vnodes = vnodes
        self.ring = []            # sorted list of hash positions
        self.owner = {}           # position -> physical node
        for n in nodes:
            self.add(n)

    @staticmethod
    def _hash(s):
        # first 8 bytes of md5 as an unsigned 64-bit int (fast enough, well spread)
        return int.from_bytes(hashlib.md5(s.encode()).digest()[:8], "big")

    def add(self, node):
        for i in range(self.vnodes):
            h = self._hash(f"{node}#{i}")
            bisect.insort(self.ring, h)
            self.owner[h] = node

    def remove(self, node):
        for i in range(self.vnodes):
            h = self._hash(f"{node}#{i}")
            self.ring.remove(h)
            del self.owner[h]

    def get(self, key):
        h = self._hash(key)
        i = bisect.bisect_right(self.ring, h) % len(self.ring)   # wrap around
        return self.owner[self.ring[i]]

ring = ConsistentHashRing(["c1", "c2", "c3"])
before = {k: ring.get(k) for k in (f"user:{i}" for i in range(10000))}
ring.add("c4")
moved = sum(1 for k, n in before.items() if ring.get(k) != n)
print(moved / 10000)   # ≈ 0.25: only keys that now belong to c4 moved

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;

public final class ConsistentHashRing {
    private final TreeMap<Long, String> ring = new TreeMap<>();
    private final int vnodes;

    public ConsistentHashRing(int vnodes) { this.vnodes = vnodes; }

    private static long hash(String s) {
        try {
            byte[] d = MessageDigest.getInstance("MD5").digest(s.getBytes(StandardCharsets.UTF_8));
            long h = 0;
            for (int i = 0; i < 8; i++) h = (h << 8) | (d[i] & 0xFF);
            return h;
        } catch (Exception e) { throw new IllegalStateException(e); }
    }

    public void addNode(String node) {
        for (int i = 0; i < vnodes; i++) ring.put(hash(node + "#" + i), node);
    }

    public void removeNode(String node) {
        for (int i = 0; i < vnodes; i++) ring.remove(hash(node + "#" + i));
    }

    public String nodeFor(String key) {
        if (ring.isEmpty()) throw new IllegalStateException("no nodes");
        SortedMap<Long, String> tail = ring.tailMap(hash(key));
        return tail.isEmpty() ? ring.firstEntry().getValue() : ring.get(tail.firstKey());
    }

    public static void main(String[] args) {
        ConsistentHashRing r = new ConsistentHashRing(150);
        for (String n : new String[]{"c1", "c2", "c3"}) r.addNode(n);
        System.out.println(r.nodeFor("user:42"));
    }
}

Note the Java version uses signed longs, so the ring's "start" is at Long.MIN_VALUE; that is fine because all comparisons use the same ordering. Lookups are O(log V) where V is total virtual nodes (a few thousand), effectively constant.

Hash slots: the Redis Cluster alternative

Redis Cluster does not use a ring. It maps every key to one of 16,384 slots via CRC16(key) mod 16384, and each primary owns a set of slots. Resharding moves whole slots between nodes. It is consistent hashing's goal (minimal movement) with an explicit, inspectable mapping. Hash tags ({user42}:cart, {user42}:profile) force related keys into the same slot so multi-key operations work. More in the Redis deep dive.

Who does the routing?

The cluster membership and slot map live in a coordination service (ZooKeeper/etcd) or are gossiped between nodes, as Redis Cluster does.

Replication

Replication buys two things: availability (a replica takes over when a primary dies, so its keys do not all miss at once) and read scaling (hot read traffic spreads over replicas).

• Leader-follower, asynchronous is standard for caches. Writes go to the primary and stream to replicas. A failover can lose the last few milliseconds of writes, which is acceptable for a cache.
• Failure detection: Redis Sentinel or Cluster's gossip marks a primary as failed after a timeout and a quorum agreement, then promotes a replica.
• Reading from replicas gives slightly stale reads. Fine for product pages, risky for "did my payment go through". Route read-your-writes paths to the primary.
• Cross-AZ placement: put each replica in a different availability zone so one zone outage does not take out a shard.

Memcached has no built-in replication; large Memcached users replicate at the client or proxy level (write to two pools) or simply accept misses on node loss and protect the database in other ways, as described below.

Eviction: LRU, LFU and TTL

When memory hits its limit, something must go. The policy determines your hit ratio.

Approximations used in practice

A precise LRU linked list costs two pointers per key and lock contention on every read. Redis instead does approximated LRU: each key stores a 24-bit last-access clock; on eviction Redis samples a few keys (default 5, configurable via maxmemory-samples) and evicts the best candidate from a small pool. With 10 samples the result is close to true LRU at a fraction of the cost.

Redis's LFU mode stores an 8-bit logarithmic (Morris-style) counter that increments probabilistically, so 255 can represent very large access counts, and decays the counter over time so yesterday's viral key eventually becomes evictable. Policies are chosen with maxmemory-policy: allkeys-lru, allkeys-lfu, volatile-lru, volatile-ttl, noeviction and others.

Memcached uses a segmented LRU per slab class (hot, warm, cold sub-queues), which resists scan pollution because a key must be hit again to be promoted out of the cold segment. Mentioning segmented LRU (or W-TinyLFU, used by Java's Caffeine library) signals depth.

Write Policies: Cache-Aside, Read-Through, Write-Through, Write-Behind

import json, random

TTL = 600

def get_product(pid, cache, db):
    key = f"product:{pid}"
    raw = cache.get(key)
    if raw is not None:
        return json.loads(raw)                       # hit
    row = db.fetch_product(pid)                      # miss → source of truth
    if row is not None:
        cache.set(key, json.dumps(row), ex=TTL + random.randint(0, 60))  # jittered TTL
    return row

def update_price(pid, price, cache, db):
    db.update_price(pid, price)                      # 1. commit to DB
    cache.delete(f"product:{pid}")                   # 2. invalidate, do not set

Cache Invalidation and the Stale-Set Race

Why delete instead of updating the cached value? Consider two writers. A sets price 100, B sets price 120. The DB commits A then B (final 120), but the cache writes arrive B then A, leaving 100 in the cache forever (until TTL). Deleting avoids ordering problems between writers.

A subtler race remains with deletes. Reader R misses and reads the old value 100 from the DB. Writer W updates the DB to 120 and deletes the key. Then R, delayed, sets the stale 100 into the cache. Fixes:

• Leases: on a miss, the cache hands the reader a lease token; a delete invalidates outstanding leases; a set with an invalidated lease is rejected. Facebook's "Scaling Memcache at Facebook" paper (NSDI 2013) describes exactly this, and also uses leases to throttle stampedes.
• Versioned values: store a version or updated_at with the value and use compare-and-set so an older version cannot overwrite a newer one.
• Delayed double delete: delete, then delete again a few hundred ms later. Crude but common.
• CDC-driven invalidation: stream the DB's change log (binlog via Debezium, Kafka) to an invalidator service, so every committed write triggers a delete even if the app forgets. This also handles writes made by other services and scripts.

And always a TTL as the final backstop.

Cache Stampede (Thundering Herd)

The home page config key is read 50,000 times a second. It expires. In the next 50 ms, a few thousand requests miss and all run the same expensive query. The database falls over, the recompute times out, the key never gets set, and the herd keeps coming. Protections, in order of how often I recommend them:

1. Request coalescing / single-flight. Within a process, concurrent misses for the same key share one in-flight load. Go's singleflight and Caffeine's loading cache do this.
2. Distributed lock on recompute. The first misser takes SET lock:key 1 NX PX 5000 and recomputes; others briefly wait and retry the cache, or get the stale value.
3. Serve stale while revalidating. Store a soft expiry inside the value and a longer hard TTL. After the soft expiry, one request refreshes in the background while everyone else gets the slightly stale value.
4. Probabilistic early expiration (XFetch). Each reader recomputes early with a probability that rises as expiry approaches, so exactly one lucky request usually refreshes before expiry and no lock is needed.
5. Pre-warming. For known hot keys (a sale banner), refresh them on a schedule so they never expire.

import math, random, time

BETA = 1.0   # > 1 favours earlier recomputation

def fetch(key, cache, recompute, ttl):
    entry = cache.get(key)          # stores (value, delta, expiry)
    now = time.time()
    if entry is not None:
        value, delta, expiry = entry
        # delta = how long the last recompute took; -log(rand) is exponentially distributed
        if now - delta * BETA * math.log(random.random()) < expiry:
            return value             # not yet our turn to refresh
    start = time.time()
    value = recompute()
    delta = time.time() - start
    cache.set(key, (value, delta, time.time() + ttl), ex=ttl)
    return value

Since log(random()) is negative, the term -delta × BETA × log(rand) is a positive random "head start". Keys that take longer to recompute (large delta) start refreshing earlier. This is from the paper "Optimal Probabilistic Cache Stampede Prevention" (Vattani, Chierichetti, Lowenstein, VLDB 2015).

The Hot Key Problem

Consistent hashing spreads keys evenly, not traffic. During a Flipkart Big Billion Days sale, one product key or one "sale config" key can receive a large fraction of all reads, pinning one shard's CPU and NIC while the other 44 idle.

Detect

• Sample commands on the server (Redis has --hotkeys in redis-cli when LFU is enabled) or count keys in the client library/proxy and report the top-K.
• Alert on per-shard skew, not just cluster averages.

Mitigate

• Local L1 cache: a tiny in-process cache (a few thousand entries, 1–5 s TTL) in each app server in front of the distributed cache. 200 app servers each fetching the hot key once per second replaces 100K reads/s with 200.
• Key replication / salting: write product:123#0 … product:123#9 to different shards; readers pick a random suffix. Writes must update (or delete) all copies.
• Read from replicas of the hot shard.
• Split big values: a hot 500 KB value is a bandwidth problem; break it up or compress.

Hot write keys (a global counter) need a different trick: shard the counter into N sub-counters and sum on read, the same approach used for rate limiter hot keys.

Failure Handling

The gutter pool idea also comes from Facebook's memcache paper: rehashing a dead node's keys onto live nodes can cascade, because the extra load from the dead node's hot keys may kill the next node too. Isolating that traffic to a dedicated, short-lived pool contains the blast radius.

The deeper principle: the database must survive a cache miss storm, at least in degraded form. Put a circuit breaker or concurrency limit on cache-miss database queries, so a cache outage produces slow pages and partial results rather than a dead database. See circuit breaker and retry patterns.

Putting It Together: Architecture

Note what this design is not: it is not a durable distributed key-value store. A Dynamo-style store adds quorum reads and writes, hinted handoff, anti-entropy and persistence because it is the source of truth. A cache can skip all of that because the database behind it is. If the interviewer pushes toward durability, call out that shift explicitly.

Interview Tips: What Gets Probed

• "What happens when you add a node?" Explain modulo vs consistent hashing with the ~91% vs ~1/N numbers.
• "Why virtual nodes?" Even load, spread of failover load, weighted capacity.
• "How do you keep cache and DB consistent?" Delete-on-write, TTLs, leases or versioning, CDC invalidation. Admit it is eventually consistent.
• "A key gets 1M reads/s." L1 cache, key salting, replicas.
• "Cache goes down. Then what?" Protect the DB: gutter pool, warm-up, miss rate limiting.
• "Redis or Memcached?" Memcached for simple, multi-threaded blob caching; Redis when you need data structures, persistence, replication or atomic scripts.

The same partitioning ideas reappear in database sharding, and the eviction logic is the LRU problem in miniature, which is why this question often shows up in both HLD and machine coding rounds.

Frequently Asked Questions

Pranjal Jain

Ex-Microsoft SDE · IIT Kanpur · Founder of Prepflix. Helps engineers crack startup and product company interviews across India.