I Build a blog API with Redis - Here's every problem I Hit

Question

I build a blog API with FastAPI + Redis + MySQL.
Three cache problems almost killed my app.
If you also face these problems!
Here's how I fixed each one.

---

1. Cache Penetration

What's Cache Penetration?

Cache Penetration refers to a scenario where the requested data exists neither in cache nor in the database,causing every request to hit the database directly,thereby increasing the database load.

How to solve?

1.Caching null values

Steps:

• When a queried key exists in neither the cache nor the database,the result (such as null or an empty value) is cached with a short TTL.
• This way,subsequent identical requests can be served directly from the cache,avoiding frequent database access.

Example:

...
redis = get_redis()
data = redis.get(cache_key)

# First, check the cache. If the cache entry exists, check whether it is __NULL__
if data is not None:
    if data == "__NULL__":
        return {"data": None, "source": "cache"}
    return {"data": data, "source": "cache"}

# Second, cache miss, query database
value = query_database(cache_key)

# Third, cache the result whether it exists in DB or not
if value is not None:
    redis.setex(cache_key, 60, value)          # cache real data
else:
    redis.setex(cache_key, 10, "__NULL__")     # cache null sentinel (short TTL)

return value

---

2.Cache breakdown

What's Cache breakdown?

Cache breakdownoccurs when a popular data entry expires in the cache while a massive number of concurrent requests are hitting that same data at the exact same time.All those requests penetrate straight through to the database,putting enormous pressure on it and potentially causing system performance issues.

How to solve?

Lock mechanism:When the cache expires,use a locking mechanism to ensure that only one thread can access the database and update the cache.The other threads wait util the cache is rebuilt before reading the data.

Example:

# Generate a lock key based on the given key
lock_key = f"mutex:{key}"

# Use SETNX operation to acquire the lock. If the key does not exist, set it with an expiration time of 300 seconds.
# This prevents concurrent access.
lock_acquired = redis_client.set(lock_key, "1", nx=True, ex=300)

if lock_acquired:
    try:
        # Successfully acquired the lock, query the database
        value = db_query_func(key)
        if value is not None:
            # Write the result to the cache with an expiration time of 3600 seconds
            redis_client.setex(key, 3600, value)
        return value
    finally:
        # Release the lock
        redis_client.delete(lock_key)
else:
    # Failed to acquire the lock, sleep for 0.1 seconds and then retry
    time.sleep(0.1)
    return get_data_with_mutex(key, redis_client, db_query_func)

---

3.Cache avalanche

What's cache avalanche?

Cache avalancerefers to a situation where a large amount of cached data expires at the same time or the cache service goes down. As a result, all requests are directly sent to the database, causing a sudden surge in the database's pressure and even leading to its downtime.

How to solve?

Randomized TTL

• The core idea is to avoid a large number of keys expiring simultaneously.
• When setting the expiration time for the cache, add a random value.

Example:

import random

expire_time = time + random.randint(0,300)
redis.set(key,value,ex=expire_time)

---

What about you?

Have you run into any of these cache problems in your projects?
Or do you use a different strategy? Let me know in the comments

If this was helpful, follow me for more backend deep dives.