التخزين المؤقت — Redis لتسريع الاستعلامات الساخنة

Q: كيف يُنفذ فلتر Bloom؟

باستخدام وحدة redisbloom أو مكتبة Python pybloom_live . تُضاف جميع معرفات المنتجات الموجودة إلى فلتر Bloom، ويُجرى فحص قبل كل استعلام. معدل الإيجابية الكاذبة حوالي 1%، وهو مقبول.

Q: كيف أراقب معدل تطابق التخزين المؤقت؟

أمر INFO stats في Redis يُرجع keyspace_hits و keyspace_misses ؛ معدل التطابق يُحسب كـ تطابقات / (تطابقات + تطابقات فاشلة). ادمجه في لوحة مراقبة Prometheus + Grafana.

التخزين المؤقت مثل الثلاجة—تحتفظ بالمكونات التي تستخدمها أكثر من غيرها (البيانات الساخنة) في الثلاجة لتكون دائمًا في متناول اليد، ولا تحتاج للذهاب إلى السوبرماركت (قاعدة البيانات) في كل مرة. لكن المكونات تنتهي صلاحيتها (TTL)، وعند إعادة التخزين تحتاج لتحديثها (إبطال التخزين المؤقت).

1. ما ستتعلمه

إدارة اتصالات Redis: دمج عميل redis-py غير المتزامن مع FastAPI
استراتيجيات التخزين المؤقت: نمط Cache-Aside، انتهاء صلاحية TTL، تصميم تسمية مفاتيح التخزين المؤقت
إبطال التخزين المؤقت: الإبطال الاستباقي خلال عمليات الكتابة؛ اتساق التخزين المؤقت خلال تحديثات الأسعار المجمّعة
الحماية من اختراق/اختراق/انهيار التخزين المؤقت: فلاتر Bloom، أقفال كشف متبادلة، TTL عشوائي
سيناريو Alice: تخزين أسعار المنتجات الشائعة مؤقتًا—انخفض زمن استجابة P99 من 200 مللي ثانية إلى 5 مللي ثانية

2. القصة الحقيقية لـ Alice

(1) نقطة الألم: الاستعلامات الشائعة تُغرق قاعدة البيانات

لدى PriceTracker 100 منتج شائع يشكل 80% من حجم الاستعلامات، بمعدل 8000 استعلام في الثانية. كل استعلام يمر عبر PostgreSQL، مما يسبب ارتفاع وحدة المعالجة المركزية لقاعدة البيانات إلى 90% وزمن استجابة P99 يبلغ 200 مللي ثانية. يقول Charlie أن إضافة مثيل قاعدة بيانات ستكلف 500 دولار شهريًا، لكن 80% من نتائج الاستعلام لن تتغير خلال 5 دقائق.

(2) حلول التخزين المؤقت بـ Redis

خزن بيانات أسعار المنتجات الشائعة مؤقتًا في Redis. للاستعلامات، تحقق من Redis أولًا (5 مللي ثانية)؛ إذا لم يكن هناك تطابق، استعلم PostgreSQL (50 مللي ثانية) واكتب النتيجة مرة أخرى إلى Redis مع TTL مدته 5 دقائق.

PYTHON

async def get_product_cached(product_id: int, db, redis):
    cached = await redis.get(f"product:{product_id}")
    if cached:
        return json.loads(cached)
    product = await db.execute(select(Product).where(Product.id == product_id))
    data = product.scalar_one_or_none()
    await redis.setex(f"product:{product_id}", 300, json.dumps(data))
    return data

(3) العائد

تصل الاستعلامات الشائعة إلى ذاكرة Redis المؤقتة بنسبة 80%؛ انخفض زمن استجابة P99 من 200 مللي ثانية إلى 5 مللي ثانية؛ انخفض QPS لقاعدة البيانات من 8000 إلى 1600؛ قلل Charlie مثيلات قاعدة البيانات من 4 إلى 1؛ توفير 1500 دولار شهريًا.

3. إدارة اتصالات Redis

(1) دمج العميل غير المتزامن

(1) ▶ مثال: اعتماد اتصال Redis

PYTHON

import redis.asyncio as aioredis
from fastapi import FastAPI, Depends

REDIS_URL = "redis://localhost:6379/0"

# تجمع اتصالات Redis على مستوى التطبيق
redis_pool = aioredis.ConnectionPool.from_url(REDIS_URL, max_connections=20)

async def get_redis() -> aioredis.Redis:
    """عميل Redis محدد النطاق للطلب من تجمع الاتصالات"""
    client = aioredis.Redis(connection_pool=redis_pool)
    try:
        yield client
    finally:
        await client.aclose()

# دورة حياة بدء/إيقاف التشغيل
app = FastAPI()

@app.on_event("startup")
async def startup():
    global redis_pool
    redis_pool = aioredis.ConnectionPool.from_url(REDIS_URL, max_connections=20)

@app.on_event("shutdown")
async def shutdown():
    await redis_pool.disconnect()

الناتج:

TEXT

# تم تعريف الدالة بنجاح

(2) اختيار هياكل بيانات Redis

الهيكل	الأمر	حالة الاستخدام	حالة استخدام PriceTracker
String	SET/GET	تخزين مؤقت أحادي القيمة	تخزين أسعار المنتجات مؤقتًا
Hash	HSET/HGET	تخزين مؤقت للكائنات	تخزين تفاصيل المنتج مؤقتًا
List	LPUSH/LRANGE	سلسلة زمنية	سجل تغييرات الأسعار
Set	SADD/SMEMBERS	مجموعة بدون تكرار	معرفات الأسعار المُدفوعة
ZSet	ZADD/ZRANGE	لوحة المتصدرين	المنتجات الأكثر مبيعًا

4. نمط Cache-Aside

(1) عملية القراءة/الكتابة

flowchart TD
    Request[طلب قراءة] --> CheckRedis{تطابق التخزين المؤقت؟}
    CheckRedis -->|نعم| Return[إرجاع البيانات المخزنة مؤقتًا]
    CheckRedis -->|لا| QueryDB[الاستعلام من قاعدة البيانات]
    QueryDB --> WriteCache[الكتابة إلى Redis مع TTL]
    WriteCache --> Return2[إرجاع البيانات]
    
    WriteRequest[طلب كتابة] --> UpdateDB[تحديث قاعدة البيانات]
    UpdateDB --> Invalidate[إبطال التخزين المؤقت]
    Invalidate --> Return3[إرجاع النجاح]

(1) ▶ مثال: استعلام منتج Cache-Aside

PYTHON

import json
from fastapi import FastAPI, Depends
from redis.asyncio import Redis

app = FastAPI()

@app.get("/api/v1/products/{product_id}")
async def get_product(
    product_id: int,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis),
):
    # الخطوة 1: التحقق من تخزين Redis المؤقت
    cache_key = f"product:{product_id}"
    cached = await redis.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # الخطوة 2: عدم تطابق التخزين المؤقت - الاستعلام من قاعدة البيانات
    repo = ProductRepository(db)
    product = await repo.get_by_id(product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    
    # الخطوة 3: الكتابة إلى التخزين المؤقت مع TTL
    product_data = ProductResponse.model_validate(product).model_dump()
    await redis.setex(cache_key, 300, json.dumps(product_data))  # TTL 5 دقائق
    
    return product_data

الناتج:

TEXT

# تم تعريف الدالة بنجاح

(2) ▶ مثال: الإبطال الاستباقي خلال عمليات الكتابة

PYTHON

@app.put("/api/v1/products/{product_id}")
async def update_product(
    product_id: int,
    update: ProductUpdate,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis),
    user=Depends(get_current_user),
):
    repo = ProductRepository(db)
    product = await repo.update(product_id, update.model_dump(exclude_unset=True))
    
    # إبطال التخزين المؤقت بعد الكتابة
    await redis.delete(f"product:{product_id}")
    
    return ProductResponse.model_validate(product).model_dump()

الناتج:

TEXT

# تم تعريف الدالة بنجاح

(2) اصطلاحات تسمية مفاتيح التخزين المؤقت

النمط	اسم العنصر	TTL	الوصف
تخزين مؤقت في الذاكرة	`product:{id}`	5 دقائق	تفاصيل منتج واحد
تخزين مؤقت للقائمة	`products:cat:{category}:p:{page}`	دقيقتان	قائمة مقسمة
عداد	`count:products:cat:{category}`	دقيقة واحدة	عدد المنتجات
الأفضل	`ranking:products:hot`	10 دقائق	ترتيب ZSet
عداد تحديد المعدل	`ratelimit:{ip}`	60 ثانية	تحديد معدل API

5. الحماية من اختراق/اختراق/انهيار التخزين المؤقت

(1) المشاكل الثلاث الرئيسية للتخزين المؤقت

المشكلة	السبب	استراتيجية التخفيف
الاختراق	البحث عن بيانات غير موجودة، تجاوز التخزين المؤقت للوصول إلى DB مباشرة	فلاتر Bloom، تخزين القيم الفارغة مؤقتًا
التحميل الزائد	تدفق ضخم من الطلبات يضرب قاعدة البيانات لحظة انتهاء صلاحية مفتاح ساخن	أقفال كشف متبادلة، مفاتيح لا تنتهي صلاحيتها أبدًا + تحديثات غير متزامنة
الانهيار	انتهاء صلاحية مفاتيح بشكل جماعي	إزاحة TTL عشوائية

(1) ▶ مثال: تخزين القيم الفارغة مؤقتًا لمنع الاختراق

PYTHON

@app.get("/api/v1/products/{product_id}")
async def get_product_with_null_cache(
    product_id: int,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis),
):
    cache_key = f"product:{product_id}"
    cached = await redis.get(cache_key)
    
    if cached:
        data = json.loads(cached)
        if data.get("_null"):
            raise HTTPException(status_code=404, detail="Product not found")
        return data
    
    # الاستعلام من DB
    repo = ProductRepository(db)
    product = await repo.get_by_id(product_id)
    
    if not product:
        # تخزين القيمة الفارغة مؤقتًا مع TTL قصير لمنع اختراق التخزين المؤقت
        await redis.setex(cache_key, 60, json.dumps({"_null": True}))
        raise HTTPException(status_code=404, detail="Product not found")
    
    product_data = ProductResponse.model_validate(product).model_dump()
    await redis.setex(cache_key, 300, json.dumps(product_data))
    return product_data

الناتج:

TEXT

# تم تعريف الدالة بنجاح

(2) ▶ مثال: منع الاختراق بأقفال كشف متبادلة

PYTHON

import asyncio

async def get_product_with_lock(
    product_id: int,
    db: AsyncSession,
    redis: Redis,
):
    cache_key = f"product:{product_id}"
    lock_key = f"lock:product:{product_id}"
    
    cached = await redis.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # محاولة الحصول على القفل (طلب واحد فقط يعيد بناء التخزين المؤقت)
    lock_acquired = await redis.set(lock_key, "1", nx=True, ex=10)
    
    if lock_acquired:
        try:
            # هذا الطلب يعيد بناء التخزين المؤقت
            repo = ProductRepository(db)
            product = await repo.get_by_id(product_id)
            if product:
                data = ProductResponse.model_validate(product).model_dump()
                await redis.setex(cache_key, 300, json.dumps(data))
                return data
        finally:
            await redis.delete(lock_key)
    else:
        # الطلبات الأخرى تنتظر قليلًا وتعيد محاولة التخزين المؤقت
        await asyncio.sleep(0.1)
        cached = await redis.get(cache_key)
        if cached:
            return json.loads(cached)
    
    raise HTTPException(status_code=404, detail="Product not found")

الناتج:

TEXT

# تم تعريف الدالة بنجاح

(3) ▶ مثال: TTL عشوائي لمنع الانهيار

PYTHON

import random

def get_cache_ttl(base_ttl: int = 300, jitter: int = 60) -> int:
    """إضافة اضطراب عشوائي إلى TTL لمنع انهيار التخزين المؤقت"""
    return base_ttl + random.randint(-jitter, jitter)

# الاستخدام
await redis.setex(cache_key, get_cache_ttl(300, 60), json.dumps(data))
# TTL: 240-360 ثانية (5 دقائق ± 1 دقيقة اضطراب)

الناتج:

TEXT

# تم تعريف الدالة بنجاح

(2) مقارنة سياسات اتساق التخزين المؤقت

الاستراتيجية	عملية الكتابة	المزايا	العيوب
Cache-Aside	تحديث DB أولًا، ثم مسح التخزين المؤقت	بسيط وموثوق	نافذة عدم اتساق قصيرة
Write-Through	الكتابة في التخزين المؤقت وDB في نفس الوقت	اتساق عالي	زمن كتابة مرتفع
Write-Behind	الكتابة في التخزين المؤقت أولًا، ثم بشكل غير متزامن إلى قاعدة البيانات	أداء كتابة جيد	احتمال فقدان البيانات

❓ أسئلة شائعة

س هل يجب أن يخزن التخزين المؤقت البيانات بتنسيق JSON أم Pickle؟

ج نوصي بـ JSON. فهو قابل للقراءة البشرية، وعبر اللغات، وآمن (Pickle لديه ثغرات إلغاء التسلسل). orjson أسرع 3-10 مرات من json.

س ما هو إعداد TTL المناسب؟

ج يعتمد على مدى تكرار تغيير البيانات. بيانات أسعار PriceTracker تُحدث كل 5 دقائق (عدم اتساق قصير مقبول)، تفاصيل المنتج كل 30 دقيقة، ولوحات المتصدرين كل 10 دقائق. يُضاف إزاحة عشوائية لمنع الانهيار.

س كيف يجب تعيين حجم تجمع اتصالات Redis؟

ج طابقه مع عدد عمال FastAPI. كل عامل يحتاج 5-10 اتصالات Redis. 4 عمال x 10 = 40 اتصال؛ عين max_connections إلى 50 للسماح ببعض الهامش.

س كيف يُنفذ فلتر Bloom؟

ج باستخدام وحدة redisbloom أو مكتبة Python pybloom_live. تُضاف جميع معرفات المنتجات الموجودة إلى فلتر Bloom، ويُجرى فحص قبل كل استعلام. معدل الإيجابية الكاذبة حوالي 1%، وهو مقبول.

س ماذا يجب أن أفعل في حالة عدم اتساق بين التخزين المؤقت وقاعدة البيانات؟

ج في وضع Cache-Aside، عدم الاتساق المؤقت طبيعي. الاستراتيجية: حدث DB أولًا، ثم احذف التخزين المؤقت (بدلاً من تحديثه)، واستخدم حذفًا مزدوجًا مؤجلًا (حذف التخزين المؤقت → تحديث DB → تأخير ثم حذف مرة أخرى) لتقليل نافذة عدم الاتساق.

س كيف أراقب معدل تطابق التخزين المؤقت؟

ج أمر INFO stats في Redis يُرجع keyspace_hits و keyspace_misses؛ معدل التطابق يُحسب كـ تطابقات / (تطابقات + تطابقات فاشلة). ادمجه في لوحة مراقبة Prometheus + Grafana.

📖 ملخص

عميل Redis غير المتزامن (redis.asyncio) + تجمع الاتصالات + حقن اعتماد yield مدمج مع FastAPI
وضع Cache-Aside: للقراءة، تحقق من التخزين المؤقت أولًا؛ إذا لم يكن هناك تطابق، استعلم قاعدة البيانات واكتب البيانات إلى التخزين المؤقت؛ للكتابة، حدث قاعدة البيانات واحذف إدخال التخزين المؤقت
مفاتيح التخزين المؤقت تُسمى وفقًا نمط {entity}:{id}، مع TTL مختلفة لسيناريوهات مختلفة
تخزين القيم الفارغة مؤقتًا لمنع الاختراق، وأقفال كشف متبادلة لمنع التحميل الزائد، وTTL عشوائي لمنع الانهيار—ثلاث استراتيجيات دفاع رئيسية
PriceTracker: معدل تطابق التخزين المؤقت للمنتجات الشائعة 80%؛ انخفض زمن استجابة P99 من 200 مللي ثانية إلى 5 مللي ثانية

📝 تمارين

تمرين أساسي (الصعوبة ⭐): قم بتكوين اعتماد عميل Redis غير المتزامن، ونفذ نقاط نهاية تخزين مؤقت GET/SET بسيطة، وتحقق من إمكانية تخزين واسترجاع البيانات. تلميح: redis.asyncio + Depends(get_redis) + setex()
تمرين متقدم (الصعوبة ⭐⭐): نفذ نمط Cache-Aside للبحث عن منتجات PriceTracker: القراءة من التخزين المؤقت → إذا عدم تطابق التخزين المؤقت، استعلم DB → اكتب إلى التخزين المؤقت (TTL 5 دقائق)؛ بعد أن تُحدث عملية الكتابة DB، احذف إدخال التخزين المؤقت. تلميح: await redis.get(key) + await redis.delete(key)
تحدي (الصعوبة ⭐⭐⭐): نفذ حماية شاملة للتخزين المؤقت—منع اختراق بتخزين القيم الفارغة (TTL قصير 60 ثانية)، ومنع تحميل زائد بقفل كشف متبادل (SET lock NX EX)، ومنع انهيار بـ TTL عشوائي (300 ثانية ± 60 ثانية)—واكتب اختبارات للتحقق من صحة منطق الحماية. تلميح: redis.set(lock_key, "1", nx=True, ex=10) + random.randint()

---|

فريق Web-Tutorial التقني

منصة دروس برمجية يديرها عدة مطورين. كل درس يتم كتابته ومراجعته بواسطة مطورين متخصصين في المجال. نعمل على ضمان دقة وموثوقية المحتوى — إذا لاحظت أي مشكلة، فيرجى إخبارنا.

من نحن

404 Not Found

التخزين المؤقت — Redis لتسريع الاستعلامات الساخنة

1. ما ستتعلمه

2. القصة الحقيقية لـ Alice

(1) نقطة الألم: الاستعلامات الشائعة تُغرق قاعدة البيانات

(2) حلول التخزين المؤقت بـ Redis

(3) العائد

3. إدارة اتصالات Redis

(1) دمج العميل غير المتزامن

(1) ▶ مثال: اعتماد اتصال Redis

(2) اختيار هياكل بيانات Redis

4. نمط Cache-Aside

(1) عملية القراءة/الكتابة

(1) ▶ مثال: استعلام منتج Cache-Aside

(2) ▶ مثال: الإبطال الاستباقي خلال عمليات الكتابة

(2) اصطلاحات تسمية مفاتيح التخزين المؤقت

5. الحماية من اختراق/اختراق/انهيار التخزين المؤقت

(1) المشاكل الثلاث الرئيسية للتخزين المؤقت

(1) ▶ مثال: تخزين القيم الفارغة مؤقتًا لمنع الاختراق

(2) ▶ مثال: منع الاختراق بأقفال كشف متبادلة

(3) ▶ مثال: TTL عشوائي لمنع الانهيار

(2) مقارنة سياسات اتساق التخزين المؤقت

❓ أسئلة شائعة

📖 ملخص

📝 تمارين