Redis Pub/Sub for Cache Invalidation: Real-Time Cross-Instance Notifications
In this tutorial, you will learn about Redis Pub/Sub for Cache Invalidation: Real. We cover key concepts, practical examples, and best practices to help you master this topic.
Redis Pub/Sub enables real-time cache invalidation by broadcasting invalidation messages to all application instances, ensuring that stale data is removed from every node's cache simultaneously when the underlying data changes.
flowchart TD
App1[App Instance 1] -->|Updates Data| DB[(Database)]
DB -->|Update Complete| App1
App1 -->|PUBLISH cache:invalidate:user:42| Redis[Redis Pub/Sub]
Redis -->|SUBSCRIBE| App1
Redis -->|SUBSCRIBE| App2[App Instance 2]
Redis -->|SUBSCRIBE| App3[App Instance 3]
App1 -->|Evict Local Cache| Local1[Local Cache]
App2 -->|Evict Local Cache| Local2[Local Cache]
App3 -->|Evict Local Cache| Local3[Local Cache]
What You'll Learn
- Pub/Sub architecture for cache invalidation
- Channel naming conventions and pattern subscriptions
- Handling missed messages with fallback TTL
- Scalability considerations for Pub/Sub invalidation
Why It Matters
Without real-time invalidation, updating data in one instance leaves other instances serving stale cached data for the entire TTL period (potentially hours). Pub/Sub invalidation reduces staleness from hours to milliseconds by broadcasting eviction messages instantly to all nodes.
Real-World Use
DodaTech's content management system uses Redis Pub/Sub for cache invalidation. When an editor updates a blog post, the publishing service broadcasts an invalidation message on channel cache:invalidate:post:{id}. All CDN edge nodes and application servers immediately evict the cached version. Readers see the updated post within 100ms of the editor clicking Publish.
Basic Pub/Sub Invalidation
Publish and subscribe for cache invalidation:
import redis
import threading
import time
import json
r = redis.Redis(decode_responses=True)
class CacheInvalidator:
def __init__(self, redis_client):
self.r = redis_client
self.listener = None
self._running = False
def publish_invalidation(self, keys):
"""Publish invalidation messages for a list of keys."""
if isinstance(keys, str):
keys = [keys]
channel = "cache:invalidate"
message = json.dumps({
"keys": keys,
"timestamp": time.time(),
"publisher": "instance_1"
})
count = self.r.publish(channel, message)
return {"channel": channel, "keys": keys, "subscribers_reached": count}
def start_listener(self, local_cache, channels=None):
"""Start a background thread to listen for invalidation messages."""
if channels is None:
channels = ["cache:invalidate"]
self._running = True
self.listener_thread = threading.Thread(
target=self._listen_loop,
args=(local_cache, channels),
daemon=True
)
self.listener_thread.start()
return {"listening": True, "channels": channels}
def _listen_loop(self, local_cache, channels):
"""Listen for invalidation messages and evict from local cache."""
pubsub = self.r.pubsub()
pubsub.subscribe(channels)
for message in pubsub.listen():
if not self._running:
break
if message["type"] != "message":
continue
try:
data = json.loads(message["data"])
for key in data["keys"]:
local_cache.pop(key, None)
print(f" Invalidated: {key}")
except json.JSONDecodeError:
continue
def stop_listener(self):
"""Stop the background listener."""
self._running = False
if self.listener_thread:
self.listener_thread.join(timeout=2)
invalidator = CacheInvalidator(r)
local_cache = {}
result = invalidator.publish_invalidation(["user:42", "post:100"])
print(f"Published invalidation: {result}")
def simulate_listener():
pubsub = r.pubsub()
pubsub.subscribe("cache:invalidate")
for msg in pubsub.listen():
if msg["type"] == "message":
data = json.loads(msg["data"])
print(f" Listener received: invalidate {data['keys']}")
return
t = threading.Thread(target=simulate_listener, daemon=True)
t.start()
time.sleep(0.1)
invalidator.publish_invalidation("post:200")
time.sleep(0.1)
Expected output:
Published invalidation: {'channel': 'cache:invalidate', 'keys': ['user:42', 'post:100'], 'subscribers_reached': 0}
Listener received: invalidate ['post:200']
Pattern-Based Invalidation
Subscribe to invalidation patterns for targeted eviction:
import redis
import threading
import time
import json
r = redis.Redis(decode_responses=True)
class PatternInvalidator:
def __init__(self, redis_client):
self.r = redis_client
self.pubsub = None
def publish_pattern(self, pattern, key):
"""Publish to a pattern-matching channel."""
channel = f"cache:{pattern}:{key}"
message = json.dumps({
"action": "invalidate",
"key": key,
"timestamp": time.time()
})
count = self.r.publish(channel, message)
return {"channel": channel, "key": key, "subscribers": count}
def listen_on_patterns(self, patterns, local_cache):
"""Subscribe to invalidation patterns and handle messages."""
self.pubsub = self.r.pubsub()
for pattern in patterns:
channel = f"cache:{pattern}:*"
self.pubsub.psubscribe(channel)
print(f" Subscribed to pattern: {channel}")
for message in self.pubsub.listen():
if message["type"] != "pmessage":
continue
channel = message["channel"]
data = json.loads(message["data"])
key = data["key"]
if data["action"] == "invalidate":
local_cache.pop(key, None)
print(f" Pattern match: evicted {key} from {channel}")
if key == "STOP":
break
invalidator = PatternInvalidator(r)
local_cache = {"user:42": "data", "post:100": "data", "config:theme": "dark"}
listener_thread = threading.Thread(
target=invalidator.listen_on_patterns,
args=(["user", "post"], local_cache),
daemon=True
)
listener_thread.start()
time.sleep(0.1)
for key in ["user:42", "post:100", "user:99"]:
prefix = key.split(":")[0]
result = invalidator.publish_pattern(prefix, key)
print(f"Published: {result['channel']}")
time.sleep(0.05)
Expected output:
Subscribed to pattern: cache:user:*
Subscribed to pattern: cache:post:*
Published: cache:user:user:42
Pattern match: evicted user:42 from cache:user:user:42
Published: cache:post:post:100
Pattern match: evicted post:100 from cache:post:post:100
Published: cache:user:user:99
Pattern match: evicted user:99 from cache:user:user:99
Reliable Invalidation with Fallback
Combine Pub/Sub with TTL-based fallback for reliability:
import redis
import time
import json
import threading
r = redis.Redis(decode_responses=True)
class ReliableInvalidator:
def __init__(self, redis_client):
self.r = redis_client
def invalidate_with_fallback(self, key, short_ttl=60):
"""Invalidate via Pub/Sub and set a short TTL as fallback."""
channel = "cache:invalidate"
message = json.dumps({
"key": key,
"short_ttl": short_ttl,
"timestamp": time.time(),
"id": f"{key}_{time.time()}"
})
subscribers = self.r.publish(channel, message)
current_ttl = self.r.ttl(key)
if current_ttl == -1 or current_ttl > short_ttl:
self.r.expire(key, short_ttl)
return {
"key": key,
"subscribers_reached": subscribers,
"fallback_ttl_set": short_ttl,
}
def process_missed_messages(self, keys):
"""Recover from potentially missed invalidation messages."""
recovered = []
for key in keys:
remaining_ttl = self.r.ttl(key)
if remaining_ttl < 60:
self.r.expire(key, remaining_ttl)
recovered.append(key)
return {"recovered": len(recovered), "keys": recovered}
invalidator = ReliableInvalidator(r)
r.setex("user:profile:42", 3600, "cached_data")
result = invalidator.invalidate_with_fallback("user:profile:42", short_ttl=30)
print(f"Invalidation with fallback: {result}")
remaining = r.ttl("user:profile:42")
print(f"TTL after fallback: {remaining}s (originally 3600s)")
Expected output:
Invalidation with fallback: {'key': 'user:profile:42', 'subscribers_reached': 0, 'fallback_ttl_set': 30}
TTL after fallback: 30s (originally 3600s)
Common Mistakes
- Relying exclusively on Pub/Sub for invalidation — if a subscriber disconnects briefly, it misses messages. Always combine Pub/Sub with TTL-based fallback to handle missed messages.
- Using too many channels — each channel creates overhead. Use channel patterns instead of creating one channel per key. A single channel with key names in the message body is more efficient.
- Not handling slow subscribers — if a subscriber processes slowly, Redis buffers messages. When the buffer limit is exceeded, messages are dropped. Use dedicated subscriber threads or processes.
- Sending large payloads through Pub/Sub — messages should be small (key names, not full data). A 1 MB message blocks Redis while it's being sent to all subscribers.
- Ignoring subscriber count — if publish returns 0 subscribers, no one received the message. Log this and ensure the TTL fallback handles the case.
Practice Questions
- How does Redis Pub/Sub help with real-time cache invalidation?
- What happens when a subscriber disconnects and reconnects?
- Why should Pub/Sub invalidation be combined with TTL-based fallback?
- What is the advantage of pattern subscriptions over channel subscriptions?
- What are the scalability limitations of Redis Pub/Sub?
Challenge
Build a multi-layer cache invalidation system that combines: (1) Redis Pub/Sub for real-time cross-instance notification, (2) local TTL fallback (set to 60s on invalidation), (3) a periodic resync that queries a "last modified" table in the database, and (4) a health check that verifies all instances received the last 10 invalidation messages. Test by disconnecting a subscriber and verifying it catches up.
FAQ
Mini Project
Build a cache invalidation dashboard that: (1) subscribes to all cache:invalidate channels, (2) tracks invalidation rate (messages/second), (3) shows the last 100 invalidated keys, (4) lists active subscribers per channel, (5) detects and alerts on messages with 0 subscribers, and (6) provides a manual invalidation form for operators. Include a reliability test that simulates subscriber disconnections.
What's Next
Continue with Geo-Distributed Caching to learn about multi-region cache topologies with Redis. Then explore Multi-Tier Caching for combining local, distributed, and CDN cache layers.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro