Using Redis as Cache Layer — ใช้ Redis เป็น Cache Layer สำหรับแอปพลิเคชัน

Categories

การใช้ Cache เป็นหนึ่งในวิธีที่มีประสิทธิภาพมากที่สุดในการเพิ่มความเร็วของแอปพลิเคชัน โดยเก็บข้อมูลที่เข้าถึงบ่อยไว้ใน Memory แทนที่จะต้องดึงจากฐานข้อมูลหรือ API ทุกครั้ง Redis เป็นตัวเลือกยอดนิยมสำหรับ Cache Layer เพราะอ่านเขียนข้อมูลได้เร็วมากในระดับ Microsecond พร้อมรองรับโครงสร้างข้อมูลหลายรูปแบบ

บทความนี้จะอธิบายแนวคิดของ Cache Layer, รูปแบบการทำ Caching ที่ใช้งานจริง, กลยุทธ์ Eviction Policy, การตั้งค่า TTL ที่เหมาะสม พร้อมตัวอย่างโค้ดสำหรับการใช้ Redis เป็น Cache ในแอปพลิเคชัน

ทำไมต้องใช้ Cache Layer

เมื่อแอปพลิเคชันเติบโตขึ้น การเข้าถึงฐานข้อมูลโดยตรงทุกครั้งจะกลายเป็นคอขวด ข้อมูลบางอย่างถูกเรียกใช้ซ้ำ ๆ เช่น รายการสินค้ายอดนิยม ข้อมูลผู้ใช้ที่ล็อกอินอยู่ หรือผลลัพธ์ของ Query ที่ซับซ้อน การเก็บข้อมูลเหล่านี้ไว้ใน Cache ช่วยลดภาระฐานข้อมูลและลดเวลาตอบสนองได้อย่างมาก

Redis เหมาะกับงาน Cache เป็นพิเศษเพราะเก็บข้อมูลใน Memory ทั้งหมดทำให้อ่านเขียนได้ในระดับ Sub-millisecond รองรับ TTL สำหรับให้ข้อมูลหมดอายุอัตโนมัติ มี Eviction Policy หลายแบบให้เลือก และรองรับการเก็บข้อมูลหลายรูปแบบทั้ง String, Hash, List และ Sorted Set

สถาปัตยกรรม Cache Layer

ในสถาปัตยกรรมทั่วไป Redis จะอยู่ระหว่างแอปพลิเคชันกับฐานข้อมูลหลัก เมื่อแอปพลิเคชันต้องการข้อมูลจะตรวจสอบที่ Cache ก่อน (Cache Hit) ถ้าพบก็ส่งกลับทันที ถ้าไม่พบ (Cache Miss) จึงไปดึงจากฐานข้อมูลแล้วเก็บใน Cache ไว้ใช้ครั้งถัดไป

# ขั้นตอนการทำงานของ Cache Layer
# 1. Client ส่ง Request
# 2. Application ตรวจสอบ Redis Cache
#    - Cache Hit → ส่งข้อมูลกลับทันที (เร็ว)
#    - Cache Miss → ดึงจาก Database → เก็บใน Cache → ส่งกลับ
# 3. ข้อมูลใน Cache หมดอายุตาม TTL → ดึงข้อมูลใหม่อีกครั้ง

Cache Patterns ที่นิยมใช้

Cache-Aside (Lazy Loading)

Cache-Aside เป็น Pattern ที่ใช้มากที่สุด แอปพลิเคชันจัดการ Cache เอง โดยตรวจสอบ Cache ก่อนทุกครั้ง ถ้าไม่พบจึงดึงจากฐานข้อมูลแล้วเขียนเข้า Cache ข้อดีคือ Cache เก็บเฉพาะข้อมูลที่ถูกเรียกใช้จริง ไม่เปลือง Memory กับข้อมูลที่ไม่มีคนเข้าถึง

import redis
import json

r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

def get_user(user_id):
    cache_key = f"user:{user_id}"

    # 1. ตรวจสอบ Cache
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)  # Cache Hit

    # 2. Cache Miss — ดึงจากฐานข้อมูล
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")

    # 3. เก็บใน Cache พร้อมตั้ง TTL 1 ชั่วโมง
    r.setex(cache_key, 3600, json.dumps(user))

    return user

Write-Through

Write-Through เขียนข้อมูลลง Cache และฐานข้อมูลพร้อมกันทุกครั้งที่มีการอัพเดต ทำให้ Cache มีข้อมูลล่าสุดเสมอ แต่การเขียนจะช้าลงเล็กน้อยเพราะต้องเขียนสองที่ เหมาะกับข้อมูลที่ต้องการความถูกต้องสูงและถูกอ่านบ่อย

def update_user(user_id, data):
    cache_key = f"user:{user_id}"

    # 1. อัพเดตฐานข้อมูล
    db.execute(f"UPDATE users SET name = '{data['name']}' WHERE id = {user_id}")

    # 2. อัพเดต Cache ทันที
    r.setex(cache_key, 3600, json.dumps(data))

    return data

Write-Behind (Write-Back)

Write-Behind เขียนข้อมูลลง Cache ก่อนแล้วค่อยเขียนลงฐานข้อมูลแบบ Asynchronous ทีหลัง ทำให้การเขียนเร็วมากจากมุมมองของแอปพลิเคชัน แต่มีความเสี่ยงที่ข้อมูลจะสูญหายถ้า Cache ล่มก่อนเขียนลงฐานข้อมูล เหมาะกับข้อมูลที่อัพเดตบ่อยมากและยอมรับการสูญหายได้บ้าง เช่น Page View Counter หรือ Analytics Data

# Write-Behind ด้วย Redis + Background Worker
def increment_page_view(page_id):
    # เขียนลง Cache ทันที (เร็วมาก)
    r.hincrby(f"pageviews:{page_id}", "count", 1)

    # Background Worker จะ sync กลับฐานข้อมูลทุก 5 นาที

# Background Worker
def sync_page_views():
    keys = r.scan_iter("pageviews:*")
    for key in keys:
        page_id = key.split(":")[1]
        count = r.hget(key, "count")
        if count:
            db.execute(f"UPDATE pages SET views = views + {count} WHERE id = {page_id}")
            r.hdel(key, "count")

Read-Through

Read-Through คล้ายกับ Cache-Aside แต่ Cache Library จัดการการดึงข้อมูลจากฐานข้อมูลให้อัตโนมัติเมื่อเกิด Cache Miss แอปพลิเคชันเรียกข้อมูลจาก Cache อย่างเดียวไม่ต้องจัดการ Miss เอง ลดความซับซ้อนของโค้ด

กลยุทธ์ Cache Invalidation

ปัญหาที่ท้าทายที่สุดของ Caching คือการทำให้ข้อมูลใน Cache ตรงกับฐานข้อมูล กลยุทธ์หลัก ๆ มีดังนี้

Time-Based Expiration (TTL)

วิธีที่ง่ายที่สุดคือตั้งเวลาหมดอายุให้ข้อมูล ข้อมูลจะถูกลบอัตโนมัติเมื่อครบเวลา แล้วจะถูกดึงใหม่จากฐานข้อมูลเมื่อถูกเรียกครั้งถัดไป เหมาะกับข้อมูลที่ยอมรับความล่าช้าในการอัพเดตได้

# ตั้ง TTL ตามประเภทข้อมูล
r.setex("product:1001", 300, product_json)    # สินค้า: 5 นาที
r.setex("user:session:abc", 3600, session)    # Session: 1 ชั่วโมง
r.setex("config:site", 86400, config_json)    # Config: 1 วัน
r.setex("search:popular", 60, results_json)   # ค้นหายอดนิยม: 1 นาที

# ตรวจสอบเวลาที่เหลือ
ttl = r.ttl("product:1001")
print(f"เหลืออีก {ttl} วินาที")

Event-Based Invalidation

ลบ Cache ทันทีเมื่อข้อมูลต้นทางเปลี่ยนแปลง ข้อมูลใน Cache จะตรงกับฐานข้อมูลเสมอ เหมาะกับข้อมูลที่ต้องการความถูกต้องสูง เช่น ราคาสินค้า หรือสถานะคำสั่งซื้อ

def update_product(product_id, new_data):
    # อัพเดตฐานข้อมูล
    db.execute("UPDATE products SET price = %s WHERE id = %s",
               (new_data['price'], product_id))

    # ลบ Cache ที่เกี่ยวข้องทั้งหมด
    r.delete(f"product:{product_id}")
    r.delete(f"category:{new_data['category_id']}:products")
    r.delete("products:bestsellers")

def delete_cache_pattern(pattern):
    """ลบ Cache ที่ตรงกับ Pattern"""
    cursor = 0
    while True:
        cursor, keys = r.scan(cursor, match=pattern, count=100)
        if keys:
            r.unlink(*keys)  # Non-blocking delete
        if cursor == 0:
            break

Version-Based Invalidation

ใช้ Version Number เป็นส่วนหนึ่งของ Cache Key เมื่อข้อมูลเปลี่ยนให้เพิ่ม Version ข้อมูลเก่าจะหมดอายุไปเอง ไม่ต้องลบ Cache Key เก่าทีละตัว

# เก็บ Version Number
r.set("product:1001:version", 1)

def get_product_cached(product_id):
    version = r.get(f"product:{product_id}:version") or "1"
    cache_key = f"product:{product_id}:v{version}"

    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    product = db.get_product(product_id)
    r.setex(cache_key, 3600, json.dumps(product))
    return product

def invalidate_product(product_id):
    # เพิ่ม Version — Cache Key เก่าจะไม่ถูกเรียกอีก
    r.incr(f"product:{product_id}:version")

Eviction Policy

เมื่อ Memory เต็ม Redis ต้องตัดสินใจว่าจะลบข้อมูลใดออก Eviction Policy กำหนดพฤติกรรมนี้ การเลือก Policy ที่เหมาะสมมีผลต่อ Cache Hit Rate โดยตรง

# ตั้ง Eviction Policy ใน redis.conf
maxmemory 2gb
maxmemory-policy allkeys-lru

# Policy ที่นิยมใช้กับ Cache:
# allkeys-lru      — ลบ Key ที่ถูกเข้าถึงนานที่สุด (แนะนำสำหรับ Cache ทั่วไป)
# allkeys-lfu      — ลบ Key ที่ถูกเข้าถึงน้อยที่สุด (Redis 4.0+)
# volatile-lru     — ลบเฉพาะ Key ที่มี TTL ตาม LRU
# volatile-lfu     — ลบเฉพาะ Key ที่มี TTL ตาม LFU
# volatile-ttl     — ลบ Key ที่ TTL เหลือน้อยที่สุดก่อน
# noeviction       — ไม่ลบ Key คืน Error เมื่อ Memory เต็ม (ไม่เหมาะกับ Cache)

# ตรวจสอบ Policy ปัจจุบัน
# redis-cli CONFIG GET maxmemory-policy

สำหรับ Cache ทั่วไป allkeys-lru เป็นตัวเลือกที่ดีเพราะลบข้อมูลที่ไม่ได้ใช้นานออกก่อน ถ้าข้อมูลมีรูปแบบการเข้าถึงที่ชัดเจน (บางข้อมูลถูกเรียกบ่อยมาก บางข้อมูลแทบไม่ถูกเรียก) allkeys-lfu จะให้ Hit Rate ดีกว่า

แนวทางการตั้ง TTL

TTL ที่เหมาะสมขึ้นอยู่กับความถี่ในการเปลี่ยนแปลงของข้อมูลและความทนทานต่อข้อมูลเก่า ข้อมูลที่เปลี่ยนแปลงบ่อยควรใช้ TTL สั้น ข้อมูลที่เปลี่ยนน้อยใช้ TTL ยาวได้ แนวทางมีดังนี้ ข้อมูล Real-time เช่น สถานะ Online หรือ Rate Limit ใช้ TTL 10 ถึง 60 วินาที ผลการค้นหาหรือ API Response ใช้ 1 ถึง 5 นาที ข้อมูลสินค้าหรือรายการทั่วไปใช้ 5 ถึง 30 นาที Session หรือ Token ใช้ 1 ถึง 24 ชั่วโมง Configuration หรือข้อมูลที่แทบไม่เปลี่ยนใช้ 1 ถึง 7 วัน

# ป้องกัน Cache Stampede ด้วย Jitter
import random

def cache_with_jitter(key, data, base_ttl):
    """เพิ่ม Random Jitter ให้ TTL เพื่อป้องกัน Key หมดอายุพร้อมกัน"""
    jitter = random.randint(0, int(base_ttl * 0.1))  # ±10%
    ttl = base_ttl + jitter
    r.setex(key, ttl, json.dumps(data))

# ตัวอย่าง: TTL 300 วินาที ± 30 วินาที
cache_with_jitter("product:1001", product_data, 300)

ป้องกันปัญหา Cache ที่พบบ่อย

Cache Stampede

Cache Stampede เกิดเมื่อ Cache Key ยอดนิยมหมดอายุพร้อมกัน ทำให้ Request จำนวนมากไปดึงข้อมูลจากฐานข้อมูลพร้อมกัน วิธีป้องกันคือใช้ Distributed Lock ให้แค่ Request แรกไปดึงข้อมูล ส่วนที่เหลือรอผลลัพธ์

import time

def get_with_lock(key, fetch_func, ttl=300):
    """ดึงข้อมูลจาก Cache พร้อม Lock ป้องกัน Stampede"""
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    lock_key = f"lock:{key}"

    # พยายามขอ Lock (ได้คนเดียว)
    if r.set(lock_key, "1", nx=True, ex=10):
        try:
            # ดึงข้อมูลจากฐานข้อมูล
            data = fetch_func()
            r.setex(key, ttl, json.dumps(data))
            return data
        finally:
            r.delete(lock_key)
    else:
        # รอจนกว่าข้อมูลจะถูกเขียนลง Cache
        for _ in range(50):
            time.sleep(0.1)
            cached = r.get(key)
            if cached:
                return json.loads(cached)

        # Fallback: ดึงจากฐานข้อมูลโดยตรง
        return fetch_func()

Cache Penetration

Cache Penetration เกิดเมื่อมีการ Request ข้อมูลที่ไม่มีอยู่จริง ทำให้ทุก Request ต้องไปถึงฐานข้อมูลเพราะไม่มี Cache ให้ตอบ วิธีป้องกันคือเก็บ Null Value ใน Cache ด้วย TTL สั้น ๆ หรือใช้ Bloom Filter กรอง Key ที่ไม่มีอยู่จริงออกก่อน

def get_user_safe(user_id):
    cache_key = f"user:{user_id}"

    cached = r.get(cache_key)
    if cached == "NULL":
        return None  # ข้อมูลนี้ไม่มีจริง (เก็บไว้ป้องกัน Penetration)
    if cached:
        return json.loads(cached)

    user = db.get_user(user_id)
    if user is None:
        # เก็บ Null Value ด้วย TTL สั้น (5 นาที)
        r.setex(cache_key, 300, "NULL")
        return None

    r.setex(cache_key, 3600, json.dumps(user))
    return user

Cache Avalanche

Cache Avalanche เกิดเมื่อ Cache Key จำนวนมากหมดอายุพร้อมกัน (เช่น ตั้ง TTL เท่ากันหมดเมื่อโหลดข้อมูลเริ่มต้น) ทำให้ Request ถาโถมไปที่ฐานข้อมูล วิธีป้องกันคือเพิ่ม Random Jitter ให้ TTL ตามที่แสดงด้านบน และใช้ Eviction Policy ที่เหมาะสม

ตัวอย่าง Cache Layer สำหรับ Web Application

Python (Flask + Redis)

import redis
import json
import hashlib
from flask import Flask, request, jsonify

app = Flask(__name__)
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)

def cache_response(prefix, ttl=300):
    """Decorator สำหรับ Cache API Response"""
    def decorator(func):
        def wrapper(*args, **kwargs):
            # สร้าง Cache Key จาก URL + Parameters
            raw_key = f"{prefix}:{request.url}:{json.dumps(request.args.to_dict())}"
            cache_key = hashlib.md5(raw_key.encode()).hexdigest()

            cached = r.get(cache_key)
            if cached:
                return jsonify(json.loads(cached))

            result = func(*args, **kwargs)
            r.setex(cache_key, ttl, json.dumps(result))
            return jsonify(result)
        wrapper.__name__ = func.__name__
        return wrapper
    return decorator

@app.route('/api/products')
@cache_response("products", ttl=300)
def get_products():
    products = db.query("SELECT * FROM products WHERE active = 1")
    return products

@app.route('/api/products/<int:product_id>')
@cache_response("product", ttl=600)
def get_product(product_id):
    product = db.query("SELECT * FROM products WHERE id = %s", (product_id,))
    return product

Node.js (Express + ioredis)

const Redis = require('ioredis');
const express = require('express');
const crypto = require('crypto');

const redis = new Redis({ host: '127.0.0.1', port: 6379 });
const app = express();

function cacheMiddleware(ttl = 300) {
  return async (req, res, next) => {
    const cacheKey = crypto
      .createHash('md5')
      .update(req.originalUrl)
      .digest('hex');

    const cached = await redis.get(cacheKey);
    if (cached) {
      return res.json(JSON.parse(cached));
    }

    // Override res.json เพื่อเก็บผลลัพธ์ลง Cache
    const originalJson = res.json.bind(res);
    res.json = (data) => {
      redis.setex(cacheKey, ttl, JSON.stringify(data));
      return originalJson(data);
    };

    next();
  };
}

app.get('/api/products', cacheMiddleware(300), async (req, res) => {
  const products = await db.query('SELECT * FROM products WHERE active = 1');
  res.json(products);
});

app.get('/api/products/:id', cacheMiddleware(600), async (req, res) => {
  const product = await db.query('SELECT * FROM products WHERE id = ?', [req.params.id]);
  res.json(product);
});

PHP (Laravel + Redis Cache)

// config/cache.php — ตั้งค่า Redis เป็น Cache Driver
// 'default' => env('CACHE_DRIVER', 'redis'),

// ใช้งานใน Controller
use Illuminate\Support\Facades\Cache;

class ProductController extends Controller
{
    public function index()
    {
        $products = Cache::remember('products:all', 300, function () {
            return Product::where('active', true)->get();
        });

        return response()->json($products);
    }

    public function show($id)
    {
        $product = Cache::remember("product:{$id}", 600, function () use ($id) {
            return Product::findOrFail($id);
        });

        return response()->json($product);
    }

    public function update(Request $request, $id)
    {
        $product = Product::findOrFail($id);
        $product->update($request->all());

        // ลบ Cache ที่เกี่ยวข้อง
        Cache::forget("product:{$id}");
        Cache::forget('products:all');

        return response()->json($product);
    }
}

ใช้ Hash สำหรับ Cache Object

แทนที่จะเก็บ Object ทั้งก้อนเป็น JSON String สามารถใช้ Hash เก็บแยก Field ได้ ข้อดีคืออัพเดตเฉพาะ Field ที่เปลี่ยนได้โดยไม่ต้องเขียนทับทั้งก้อน ประหยัดทั้ง Bandwidth และลดปัญหา Race Condition

# เก็บข้อมูลสินค้าแบบ Hash
r.hset("product:1001", mapping={
    "name": "Mechanical Keyboard",
    "price": "2990",
    "stock": "150",
    "category": "peripherals"
})
r.expire("product:1001", 600)

# อ่านเฉพาะ Field ที่ต้องการ
price = r.hget("product:1001", "price")
name_and_price = r.hmget("product:1001", "name", "price")

# อัพเดตเฉพาะ Stock (ไม่กระทบ Field อื่น)
r.hincrby("product:1001", "stock", -1)

# อ่านทั้ง Object
product = r.hgetall("product:1001")

Monitoring Cache Performance

การติดตาม Cache Performance ช่วยให้รู้ว่า Cache ทำงานได้ดีแค่ไหนและควรปรับปรุงตรงไหน ตัวชี้วัดที่สำคัญ ได้แก่ Hit Rate, Memory Usage, Eviction Count และ Latency

# ดูสถิติ Cache จาก Redis INFO
redis-cli INFO stats | grep keyspace
# keyspace_hits:1234567      — จำนวน Cache Hit
# keyspace_misses:12345      — จำนวน Cache Miss

# คำนวณ Hit Rate
# Hit Rate = keyspace_hits / (keyspace_hits + keyspace_misses) * 100
# Hit Rate ที่ดีควรอยู่ที่ 90% ขึ้นไป

# ดู Memory Usage
redis-cli INFO memory | grep used_memory_human
# used_memory_human:256.50M

# ดูจำนวน Key ที่ถูก Evict
redis-cli INFO stats | grep evicted_keys
# evicted_keys:0   — ถ้าเลขสูงอาจต้องเพิ่ม maxmemory

# ดู Latency
redis-cli --latency
# min: 0, max: 1, avg: 0.15 (มิลลิวินาที)

สรุป

Redis เป็น Cache Layer ที่มีประสิทธิภาพสูง รองรับหลาย Pattern ทั้ง Cache-Aside, Write-Through และ Write-Behind สิ่งสำคัญคือการเลือก Pattern, TTL และ Eviction Policy ที่เหมาะสมกับลักษณะข้อมูล การป้องกัน Cache Stampede, Penetration และ Avalanche ก็เป็นเรื่องที่ควรวางแผนตั้งแต่เริ่มต้น และการ Monitor Hit Rate อย่างสม่ำเสมอจะช่วยให้ปรับจูนระบบได้อย่างต่อเนื่อง

แนะนำบริการ DE

การรัน Redis Cache Layer ในระดับ Production ต้องการเซิร์ฟเวอร์ที่มี RAM เพียงพอและ Network Latency ต่ำ Cloud VPS ของ DE รองรับการเลือก RAM ได้ตามขนาดข้อมูลที่ต้อง Cache พร้อม SSD NVMe ที่ช่วยให้ RDB Snapshot เร็ว เหมาะสำหรับรัน Redis ทั้ง Standalone และ Cluster

สำหรับผู้ที่ต้องการโฮสต์เว็บแอปพลิเคชันพร้อม Redis Cache โดยไม่ต้องจัดการเซิร์ฟเวอร์เอง Cloud Hosting ของ DE เป็นทางเลือกที่สะดวก รองรับ Redis Object Cache สำหรับ WordPress และเว็บแอปพลิเคชันทั่วไป