How to Design a System for 1 Million Requests per Second

"Design a system that handles a million requests per second" is a favourite senior interview question. It's a test of whether you can reason about numbers, bottlenecks and trade-offs, not whether you can name every AWS service.

The short answer: no single machine, database or cache survives 1M req/s. You get there by removing work (caching at every layer), spreading work (horizontal scaling and sharding) and deferring work (queues) — and by protecting the system when it's overloaded.

Let's build it layer by layer.

1. Do the capacity math first

Never draw boxes before you know the numbers.

Daily volume. 1,000,000 req/s × 86,400 s = 86.4 billion requests per day.

Requests in flight (Little's Law). Concurrency = throughput × latency. With a 50 ms average response time: 1,000,000 × 0.05 s = 50,000 requests in flight at any moment. Every one of them holds a connection, a thread (or a coroutine / virtual thread) and some memory.

CPU is usually the real limit. If each request needs just 5 ms of CPU, that's 1,000,000 × 5 ms = 5,000 CPU-seconds every second — roughly 5,000 cores busy all the time. Cut CPU per request to 1 ms and you need about 1,000. This is why the biggest wins come from not doing the work at all.

Bandwidth. With 10 KB responses: 1,000,000 × 10 KB = 10 GB/s ≈ 80 Gbps of egress. That alone pushes you towards a CDN.

Server count. If one tuned instance handles about 10,000 simple requests/s, you need 100 instances for the traffic that reaches your origin — then add headroom for peaks, deployments and a lost availability zone (typically run at 50–60% utilisation).

Ask about the read/write ratio. A 1M req/s system that is 95% reads (a product catalogue) is a caching problem. One that is 50% writes (an ad-click tracker) is a partitioning and streaming problem. The design is very different.

2. The request path, end to end

Users
  │
  ▼
GeoDNS / Anycast ──► nearest region
  │
  ▼
CDN edge (static + cacheable responses)
  │  only cache misses and dynamic calls continue
  ▼
L4/L7 load balancers (TLS, health checks, routing)
  │
  ▼
API gateway (auth, rate limits, request limits)
  │
  ▼
Stateless app servers (autoscaled)
  │         │                │
  ▼         ▼                ▼
Local     Redis          Kafka ──► workers
cache     cluster                     │
  │         │                         ▼
  └────► Sharded database (primary + read replicas per shard)

Each layer has one job: absorb as much traffic as possible so the next layer sees less.

3. Edge: DNS, CDN and load balancers

GeoDNS or anycast sends users to the nearest region, cutting latency and splitting traffic across regions.

The CDN serves static assets and any response that can be cached publicly: product pages, public API responses, images. For read-heavy public content, the CDN can absorb most of the traffic.

Be careful with the "80% at the CDN" claims you'll see online: it only holds for cacheable traffic. Personalised responses (your cart, your feed, your balance) can't be served from a shared edge cache, so they reach your origin no matter what.

Load balancers terminate TLS, run health checks and spread requests across instances. Use Layer 7 (HTTP) balancing for path-based routing and Layer 4 for raw throughput. Load balancers themselves scale horizontally — managed ones do this for you.

4. Keep app servers stateless

Horizontal scaling only works if any instance can serve any request:

• No in-memory sessions — keep session state in a token or in Redis.
• No local files that other instances need.
• Configuration and secrets come from outside the instance.

Then you can add or remove instances freely, and a crashed instance loses nothing.

In Java, high concurrency with blocking I/O used to mean huge thread pools. With Java 21 virtual threads, you can keep simple blocking code and still hold tens of thousands of in-flight requests per instance:

# Spring Boot 3.2+
spring:
  threads:
    virtual:
      enabled: true

5. Cache at every layer

A hit at any layer means the layers below it do no work at all. Use more than one level:

1. CDN — public, cacheable responses.
2. In-process cache (Caffeine) — hottest keys, nanosecond access, no network hop.
3. Distributed cache (Redis cluster) — shared across all instances.
4. Database — the source of truth, hit only on a miss.

With Spring's cache abstraction:

@Service
public class ProductService {

    private final ProductRepository repo;

    public ProductService(ProductRepository repo) {
        this.repo = repo;
    }

    // First call loads from the DB; later calls come from the cache.
    // sync = true lets only one thread load a missing key.
    @Cacheable(cacheNames = "products", key = "#id", sync = true)
    public ProductDto getProduct(long id) {
        return repo.findById(id)
                   .map(ProductDto::from)
                   .orElseThrow(() -> new ProductNotFoundException(id));
    }

    @CacheEvict(cacheNames = "products", key = "#id")
    public void updateProduct(long id, UpdateProductRequest req) {
        // update the DB first, then the entry is evicted
    }
}

The problems that actually bite at scale:

• Cache stampede. A hot key expires and thousands of requests hit the database at once. Fix with request coalescing (sync = true above, per instance), a short lock in Redis, or refreshing hot keys before they expire.
• Synchronised expiry. If you cache a million keys with the same TTL at the same moment, they all expire together. Add random jitter to TTLs (e.g. 10 minutes ± 1 minute).
• Hot keys. One celebrity product can overload a single Redis shard. Put it in the local cache too, or replicate the key.
• Missing data. Cache "not found" results briefly (negative caching), or a flood of requests for non-existent IDs goes straight to the database.
• Staleness. Decide per data type how stale is acceptable. Prices and stock need short TTLs or explicit invalidation; a product description can live for hours.

6. Shard the database

A single database primary handles thousands to tens of thousands of writes per second — not a million. So you split data across shards, each an independent database with its own primary and read replicas.

Choosing the shard key is the most important decision:

• It should spread load evenly (user ID usually does; country usually doesn't).
• Most queries should need only one shard (all of a user's data lives together).
• Avoid keys that create hot shards, such as timestamps, where all new writes land on the latest shard.

Modulo hashing vs consistent hashing

The simplest routing is shard = hash(key) % N. It distributes well, but it has a serious flaw: when you change N (say 16 → 17 shards), almost every key moves to a different shard, which means a massive data migration.

Consistent hashing fixes this. Shards are placed on a hash ring (each several times, as "virtual nodes"), and a key belongs to the first shard clockwise from its hash. Adding a shard only moves the keys in its slice of the ring — roughly 1/N of the data.

public class ConsistentHashRouter {

    private final TreeMap<Long, String> ring = new TreeMap<>();
    private final int virtualNodes;

    public ConsistentHashRouter(List<String> shards, int virtualNodes) {
        this.virtualNodes = virtualNodes;
        shards.forEach(this::addShard);
    }

    public void addShard(String shard) {
        for (int i = 0; i < virtualNodes; i++) {
            ring.put(hash(shard + "#" + i), shard);
        }
    }

    public String shardFor(String key) {
        Map.Entry<Long, String> e = ring.ceilingEntry(hash(key));
        return (e != null ? e : ring.firstEntry()).getValue(); // wrap around
    }

    private static long hash(String s) {
        // A stable hash (same result on every machine and restart).
        // Never use a per-process randomized hash for routing.
        byte[] d = sha256(s.getBytes(StandardCharsets.UTF_8));
        return ByteBuffer.wrap(d).getLong();
    }

    private static byte[] sha256(byte[] input) {
        try {
            return MessageDigest.getInstance("SHA-256").digest(input);
        } catch (NoSuchAlgorithmException ex) {
            throw new IllegalStateException(ex);
        }
    }
}

Two details interviewers love:

• The hash must be stable across processes. Some languages randomize string hashing per process; if you route with that, every instance sends the same user to a different shard.
• Many real systems use a lookup table instead — a mapping from key ranges (or buckets) to shards, stored in a config service. It's more flexible for rebalancing than pure hashing.

Read replicas handle read traffic for each shard. Writes go to the primary; reads can go to replicas, but replicas lag. A user who just updated their profile and reads it back from a replica may see the old value. Route "read your own writes" to the primary for a short window, or read from the cache you just updated.

Also plan for cross-shard queries (reports, search): don't run them against the shards. Stream changes into a separate analytics store or search index instead.

7. Move heavy work off the request path

If a request triggers slow work — sending emails, generating a report, image processing, calling slow partners — don't do it while the user waits. Accept the request, put a message on a queue, and respond immediately with 202 Accepted.

@RestController
@RequestMapping("/api/reports")
public class ReportController {

    private final KafkaTemplate<String, ReportJob> kafka;
    private final JobStatusStore statusStore;

    public ReportController(KafkaTemplate<String, ReportJob> kafka, JobStatusStore statusStore) {
        this.kafka = kafka;
        this.statusStore = statusStore;
    }

    @PostMapping
    public ResponseEntity<JobAccepted> create(@Valid @RequestBody ReportRequest req,
                                              @RequestHeader("Idempotency-Key") String key) {
        String jobId = statusStore.createIfAbsent(key);        // same key → same job
        kafka.send("report-jobs", jobId, ReportJob.from(jobId, req));
        return ResponseEntity.accepted()
                .location(URI.create("/api/reports/" + jobId))  // client polls this
                .body(new JobAccepted(jobId, "QUEUED"));
    }
}

Workers consume from Kafka at their own pace and update the job status. Benefits:

• Spikes are absorbed by the queue instead of overloading your services.
• Workers scale independently of the API.
• Failures are retried without the user resending anything.

Things to get right: consumers must be idempotent (messages can be delivered more than once), failing messages go to a dead-letter topic after limited retries, and you monitor consumer lag. Kafka throughput scales with the number of partitions. (More in Kafka for Backend Engineers.)

8. Protect the system: rate limits, timeouts, circuit breakers, load shedding

At this scale something is always failing. The goal is to make sure one failure doesn't take everything down.

Rate limiting at the gateway stops one client from consuming everyone's capacity. It must be distributed — an in-memory limiter on each instance lets a client get N× the limit across N instances. Use Redis with an atomic token-bucket script, and return 429 Too Many Requests with a Retry-After header. (Full design in Designing a Rate Limiter.)

Timeouts on every remote call. Without them, a slow dependency makes threads pile up until your service stops responding too.

Circuit breakers stop calling a dependency that is failing and fail fast instead, giving it time to recover. With Resilience4j in Spring Boot:

resilience4j:
  circuitbreaker:
    instances:
      pricing:
        sliding-window-size: 50
        failure-rate-threshold: 50        # open at 50% failures
        slow-call-duration-threshold: 500ms
        wait-duration-in-open-state: 10s  # then half-open and probe
  timelimiter:
    instances:
      pricing:
        timeout-duration: 300ms

@CircuitBreaker(name = "pricing", fallbackMethod = "cachedPrice")
@TimeLimiter(name = "pricing")
public CompletableFuture<Price> livePrice(long productId) {
    return pricingClient.fetch(productId);
}

private CompletableFuture<Price> cachedPrice(long productId, Throwable ex) {
    return CompletableFuture.completedFuture(priceCache.lastKnown(productId));
}

Load shedding. When you're over capacity, rejecting some requests quickly (503) is better than accepting everything and timing out for everyone. Prioritise: checkout beats recommendations.

Retries with exponential backoff and jitter — and only for idempotent operations. Naive retries multiply traffic exactly when a service is struggling. (See Resilient Microservices.)

9. Autoscale on the right signals

Scaling by CPU alone is a classic mistake. An I/O-bound service can be slow at 30% CPU because it's waiting on a database or a connection pool. Scale on signals that reflect load:

• Requests per second per instance
• p95/p99 latency
• Queue depth / consumer lag for workers

In Kubernetes, the Horizontal Pod Autoscaler can use custom metrics (for example via Prometheus Adapter or KEDA for Kafka lag). Also:

• Scale up fast, scale down slowly to avoid flapping.
• Pre-scale for known peaks (sales, launches) — autoscaling takes minutes, a traffic spike takes seconds.
• Remember that scaling app servers multiplies database connections. Put a connection pooler or proxy in front of the database, and size pools so 200 instances don't open 20,000 connections.

10. Observe everything

You can't operate what you can't see:

• Metrics: the golden signals — latency (p50/p95/p99), traffic, errors and saturation — per service and per endpoint.
• Distributed tracing (OpenTelemetry) to see which hop is slow in a request that crosses ten services.
• Structured logs with trace IDs, sampled at this volume — logging every request at 1M req/s is itself a big-data problem.
• Alerts on user-visible symptoms (error rate, latency SLOs), not on every CPU blip.

11. Survive failures: multiple zones and regions

• Run every tier across at least three availability zones, with enough headroom to lose one.
• For global traffic, run multiple regions. Active-active gives the best latency and resilience, but you must decide how data is replicated between regions and how conflicts are resolved.
• Load test before you need it, and practise failure (kill instances, zones, dependencies) so recovery is routine.

The checklist

Most companies never need 1M req/s — and the ones that do got there gradually. The right approach is the same at any scale: measure, find the real bottleneck, fix it, repeat.

Follow-up questions this topic invites — and their answers

Q: How many requests are "in flight" at 1M req/s with 50 ms latency? A: By Little's Law, concurrency = throughput × latency = 1,000,000 × 0.05 s = 50,000. That number sizes your connection pools, thread capacity and memory.

Q: Why not just buy a bigger server? A: Vertical scaling hits hard limits in CPU, memory, network and disk I/O, costs grow faster than capacity, and one machine is a single point of failure. Horizontal scaling adds capacity linearly and survives failures.

Q: Can a CDN absorb most of the traffic? A: Only the cacheable part — static assets and public, non-personalised responses. Personalised or write traffic always reaches your origin, so the CDN's share depends entirely on your workload.

Q: What's wrong with hash(key) % N for sharding? A: Changing N remaps almost every key, forcing a huge migration. Consistent hashing (with virtual nodes) or a lookup table moves only about 1/N of the keys when you add a shard.

Q: How do you choose a shard key? A: Pick a key with high cardinality that spreads load evenly and keeps most queries on a single shard, such as user ID. Avoid monotonically increasing keys like timestamps, which create a hot shard.

Q: What is a cache stampede and how do you prevent it? A: When a hot key expires, many requests miss at once and all hit the database. Prevent it with request coalescing or a short lock so only one request reloads, early background refresh, and TTL jitter.

Q: How do you handle read-after-write with read replicas? A: Replicas lag behind the primary, so read a user's own recent writes from the primary (for a short window after writing) or from a cache updated on write.

Q: Why return 202 Accepted for heavy operations? A: It decouples the user's request from slow work: the API stays fast, spikes queue up instead of overloading services, and workers scale and retry independently. The client polls a status URL or gets a webhook.

Q: Why can't each instance rate-limit in memory? A: Each instance only sees its own traffic, so a client spread across N instances gets N times the limit. A shared store like Redis with atomic operations gives one global count.

Q: What should autoscaling be based on? A: Signals that reflect load and user experience — requests per instance, latency percentiles, queue depth or consumer lag — not CPU alone. Scale up quickly, scale down slowly, and pre-scale for known events.

Q: What breaks first when you add 10× more app servers? A: Often the database: every instance opens its own connection pool, so connections multiply. Use a connection pooler/proxy, cap pool sizes and push reads to caches and replicas.

Q: How would you design this for multiple regions? A: Route users to the nearest region with GeoDNS or anycast, keep each region able to serve on its own, and choose a data strategy — per-region data ownership (users pinned to a home region) or multi-region replication with explicit conflict handling.