Bối cảnh: Khi database trở thành nút thắt cổ chai
Mình từng gặp một vấn đề khá điển hình: script Python xử lý 100K records, mỗi record cần query database để lấy thông tin cấu hình. Kết quả? Script chạy mất hơn 20 phút, database CPU nhảy lên 90%. Vấn đề không phải ở logic xử lý mà ở chỗ cùng một query được gọi đi gọi lại hàng nghìn lần với kết quả giống hệt nhau.
Redis và redis-py là thứ đã cứu cái script đó. Thay vì mỗi lần đọc dữ liệu đều phải đánh thức database, Redis giữ kết quả trong memory — tốc độ đọc/ghi tính bằng microsecond thay vì millisecond. Script 20 phút rút xuống còn 90 giây sau khi thêm cache layer. Không phải nhờ viết lại thuật toán, mà chỉ nhờ tránh làm đi làm lại cùng một việc.
Bài này tập trung vào 3 kỹ thuật cụ thể mà mình dùng thường xuyên nhất với redis-py: Caching (tránh query thừa), Pipeline (gom batch lệnh giảm round-trip network), và Distributed Lock (đảm bảo tính nhất quán khi nhiều process cùng chạy song song).
Cài đặt redis-py
Trước tiên cần có Redis server đang chạy. Cách nhanh nhất là dùng Docker:
docker run -d --name redis-local -p 6379:6379 redis:7-alpineTiếp theo, cài thư viện Python:
pip install redis
# Thêm hiredis để parse response nhanh hơn
pip install redis[hiredis]hiredis là C extension giúp parse response nhanh hơn, không có lý do gì để bỏ qua. Kiểm tra kết nối:
import redis
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
print(r.ping()) # True nếu kết nối thành côngLưu ý decode_responses=True — thiếu tham số này thì mọi giá trị trả về đều là bytes. Mình hay quên cái này lúc mới bắt đầu dùng và mất 10 phút debug lý do tại sao string comparison không hoạt động.
Cấu hình chi tiết
1. Caching với Cache-Aside Pattern
Ý tưởng khá trực quan: đọc từ Redis trước, miss thì mới xuống DB — rồi lưu kết quả lên Redis kèm TTL để lần sau khỏi xuống nữa.
import redis
import json
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
def get_user_config(user_id: int) -> dict:
cache_key = f"user:config:{user_id}"
# Đọc từ cache trước
cached = r.get(cache_key)
if cached:
return json.loads(cached)
# Cache miss — đọc từ DB
config = fetch_from_database(user_id) # Hàm query DB thực tế của bạn
# Lưu vào cache, TTL 5 phút
r.setex(cache_key, 300, json.dumps(config))
return configVới 100K records mà chỉ có 500 unique user_id, số query DB giảm từ 100K xuống còn ~500 — 200 lần ít hơn. Database CPU từ 90% xuống khoảng 15%, script từ 20 phút còn 90 giây. Thêm vài chục dòng code, không cần refactor gì.
Nếu không muốn viết boilerplate cho từng function, có thể bọc bằng decorator:
from functools import wraps
def redis_cache(ttl=300):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
key = f"cache:{func.__name__}:{args}:{sorted(kwargs.items())}"
result = r.get(key)
if result:
return json.loads(result)
result = func(*args, **kwargs)
r.setex(key, ttl, json.dumps(result))
return result
return wrapper
return decorator
@redis_cache(ttl=600)
def get_product_details(product_id: int) -> dict:
# Query database...
pass2. Pipeline — Giảm Round-trip Network
Mỗi lệnh Redis gửi đi là một round-trip network. Nếu cần chạy 1000 lệnh riêng lẻ trên mạng có latency 1ms, mất 1 giây chỉ cho overhead mạng. Pipeline gom nhiều lệnh thành một batch, gửi một lần, nhận kết quả một lần.
import redis
import time
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# Cách KHÔNG dùng pipeline (chậm)
def set_scores_naive(scores: dict):
start = time.perf_counter()
for user_id, score in scores.items():
r.set(f"score:{user_id}", score)
print(f"Naive: {time.perf_counter() - start:.3f}s")
# Cách dùng Pipeline (nhanh hơn 8-12x)
def set_scores_pipeline(scores: dict):
start = time.perf_counter()
with r.pipeline(transaction=False) as pipe: # transaction=False cho throughput cao hơn
for user_id, score in scores.items():
pipe.set(f"score:{user_id}", score)
pipe.execute() # Gửi tất cả một lần
print(f"Pipeline: {time.perf_counter() - start:.3f}s")
# Test với 10,000 records
test_scores = {f"user_{i}": i * 10 for i in range(10000)}
set_scores_naive(test_scores) # ~2.5s trên localhost
set_scores_pipeline(test_scores) # ~0.3s trên localhostPipeline cũng hữu dụng khi đọc nhiều key cùng lúc:
def get_multiple_users(user_ids: list) -> dict:
with r.pipeline(transaction=False) as pipe:
for uid in user_ids:
pipe.get(f"user:config:{uid}")
results = pipe.execute()
return {
uid: json.loads(res) if res else None
for uid, res in zip(user_ids, results)
}transaction=False tắt MULTI/EXEC wrapping để tăng throughput. Chỉ bật transaction=True khi thực sự cần atomic execution — chẳng hạn khi đọc rồi ghi dựa trên giá trị vừa đọc.
3. Distributed Lock — Tránh Race Condition
Tình huống thực tế: 3 worker process cùng check “đã gửi email chưa?” và cùng quyết định “chưa, mình gửi”. Kết quả là user nhận 3 email giống nhau. Distributed lock ngăn chuyện đó bằng cách đảm bảo chỉ một process thực thi đoạn code tại một thời điểm.
import redis
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
def send_notification_once(user_id: int, message: str) -> bool:
lock_key = f"lock:notification:{user_id}"
# Thử lấy lock với timeout 10 giây
lock = r.lock(lock_key, timeout=10, blocking_timeout=5)
acquired = lock.acquire(blocking=True)
if not acquired:
print(f"Không lấy được lock cho user {user_id}, bỏ qua")
return False
try:
# Double-check sau khi có lock
sent_key = f"notification:sent:{user_id}"
if r.exists(sent_key):
return False
# Gửi notification
send_email(user_id, message)
# Đánh dấu đã gửi, TTL 1 giờ
r.setex(sent_key, 3600, "1")
return True
finally:
lock.release() # LUÔN release trong finallylock.release() phải nằm trong finally — không có ngoại lệ. Exception mà không release thì tất cả process khác bị block cho đến khi timeout tự expire. Hai tham số cần nhớ: timeout là thời gian lock tự expire nếu process crash (safety net), còn blocking_timeout là thời gian tối đa một process chờ để lấy được lock.
Trường hợp hay gặp khác là cron job chạy song song:
def run_scheduled_job(job_name: str):
lock_key = f"job:running:{job_name}"
lock = r.lock(lock_key, timeout=300) # Job tối đa 5 phút
if not lock.acquire(blocking=False): # Non-blocking: skip nếu đang chạy
print(f"Job {job_name} đang chạy ở worker khác, skip")
return
try:
execute_job(job_name)
finally:
lock.release()Kiểm tra & Monitoring
Cache hit rate
Redis tracking hit/miss ngay trong memory, không cần external tool. Lệnh sau cho số liệu ngay lập tức:
redis-cli info stats | grep -E "keyspace_hits|keyspace_misses"
# Kết quả mẫu:
# keyspace_hits:45231
# keyspace_misses:1823
# Hit rate = 45231 / (45231 + 1823) ≈ 96%def get_cache_stats() -> dict:
info = r.info('stats')
hits = info['keyspace_hits']
misses = info['keyspace_misses']
total = hits + misses
hit_rate = (hits / total * 100) if total > 0 else 0
return {
'hits': hits,
'misses': misses,
'hit_rate': f"{hit_rate:.1f}%",
}
def get_memory_usage() -> dict:
info = r.info('memory')
return {
'used': info['used_memory_human'],
'fragmentation_ratio': info['mem_fragmentation_ratio'],
}Slow log và cấu hình memory
# Bật slow log — log lệnh chậm hơn 1ms
redis-cli config set slowlog-log-slower-than 1000 # microseconds
redis-cli slowlog get 10
# Giới hạn memory, tự evict key cũ theo LRU
redis-cli config set maxmemory 512mb
redis-cli config set maxmemory-policy allkeys-lruPolicy allkeys-lru phù hợp với hầu hết use case caching: Redis tự evict key ít được dùng nhất khi đầy bộ nhớ, thay vì trả lỗi OOM. Đặt maxmemory luôn là bước không thể bỏ qua trước khi đưa lên production.
Benchmark pipeline thực tế
import time
def benchmark_pipeline(n=5000):
test_data = {f"key:{i}": f"value:{i}" for i in range(n)}
start = time.perf_counter()
for k, v in test_data.items():
r.set(k, v)
t_normal = time.perf_counter() - start
start = time.perf_counter()
with r.pipeline(transaction=False) as pipe:
for k, v in test_data.items():
pipe.set(k, v)
pipe.execute()
t_pipeline = time.perf_counter() - start
print(f"Normal: {t_normal:.3f}s")
print(f"Pipeline: {t_pipeline:.3f}s")
print(f"Speedup: {t_normal / t_pipeline:.1f}x")
benchmark_pipeline()Trên localhost (Redis 7, Python 3.11), 5000 operations nhanh hơn 8–12x. Nhưng khi kết nối qua network thật với latency 10–50ms, cách biệt còn lớn hơn nhiều — 1000 lệnh × 10ms = 10 giây overhead. Với pipeline, cả 1000 lệnh đó vẫn chỉ là một round-trip.
Ba kỹ thuật này độc lập nhau — dùng cái nào tùy vấn đề đang gặp. Caching khi thấy database CPU cao bất thường. Pipeline khi cần ghi/đọc hàng loạt. Distributed lock khi có nhiều worker cùng xử lý một luồng dữ liệu. Áp dụng đúng chỗ, redis-py ít khi khiến bạn thất vọng.
