Redis ile Yüksek Performanslı Cache ve Message Queue

Redis Nedir ve Neden Kullanılır?

Redis (Remote Dictionary Server), açık kaynaklı, bellek içi (in-memory) çalışan bir veri yapısı deposudur. 2009 yılında Salvatore Sanfilippo tarafından geliştirilen Redis, günümüzde en popüler NoSQL çözümlerinden biri haline gelmiştir. Temel olarak anahtar-değer (key-value) yapısında çalışan Redis; string, hash, list, set, sorted set, stream gibi zengin veri yapılarını destekler.

Redis'in öne çıkan özellikleri şunlardır:

• Bellek içi çalışma: Tüm veriler RAM'de tutulduğu için mikrosaniye seviyesinde okuma/yazma performansı sunar.
• Persistence desteği: RDB snapshot ve AOF (Append Only File) mekanizmalarıyla veriler diske yazılabilir.
• Replikasyon: Master-replica mimarisiyle yüksek erişilebilirlik sağlar.
• Cluster desteği: Yatay ölçekleme ile büyük veri setlerini dağıtık olarak yönetebilir.
• Pub/Sub ve Streams: Mesajlaşma altyapısı olarak kullanılabilir.

Cache Stratejileri ve Redis

Önbellekleme (caching), modern yazılım mimarilerinin vazgeçilmez bir parçasıdır. Veritabanı sorgularının, API çağrılarının veya hesaplama yoğun işlemlerin sonuçlarını önbellekte tutarak uygulamanızın yanıt süresini dramatik şekilde iyileştirebilirsiniz. Redis, bu alanda en çok tercih edilen araçların başında gelir.

Cache-Aside (Lazy Loading) Stratejisi

En yaygın kullanılan cache stratejisidir. Uygulama önce cache'e bakar; veri varsa doğrudan döner (cache hit), yoksa veritabanından okur ve cache'e yazar (cache miss).

import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

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

    # Önce cache'e bak
    cached = r.get(cache_key)
    if cached:
        print("Cache hit!")
        return json.loads(cached)

    # Cache miss - veritabanından oku
    print("Cache miss, veritabanından okunuyor...")
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")

    # Cache'e yaz, 5 dakika TTL ile
    r.setex(cache_key, 300, json.dumps(user))

    return user

Write-Through Stratejisi

Bu stratejide veri yazma işlemi hem cache'e hem de veritabanına eş zamanlı olarak yapılır. Veri tutarlılığı yüksektir ancak yazma işlemleri biraz daha yavaştır.

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

    # Veritabanını güncelle
    db.execute("UPDATE users SET name=%s WHERE id=%s", (data['name'], user_id))

    # Cache'i de güncelle
    r.setex(cache_key, 300, json.dumps(data))

    return data

Write-Behind (Write-Back) Stratejisi

Veri önce cache'e yazılır, ardından asenkron olarak veritabanına aktarılır. Yazma performansı çok yüksektir ancak veri kaybı riski taşır. Yoğun yazma işlemlerinin olduğu senaryolarda tercih edilir.

TTL ve Eviction Politikaları

Cache yönetiminde TTL (Time To Live) ve bellek dolduğunda hangi anahtarların silineceğini belirleyen eviction politikaları kritik öneme sahiptir.

# TTL ile anahtar oluşturma
r.setex("session:abc123", 3600, "user_data")  # 1 saat
r.set("config:feature_flags", "...", ex=86400)  # 24 saat

# Mevcut bir anahtara TTL ekleme
r.expire("user:42", 600)  # 10 dakika

# Kalan süreyi sorgulama
ttl = r.ttl("session:abc123")
print(f"Kalan süre: {ttl} saniye")

Redis'in desteklediği başlıca eviction politikaları:

• noeviction: Bellek dolduğunda yazma işlemlerini reddeder.
• allkeys-lru: Tüm anahtarlar arasında en az kullanılanı siler. Genel amaçlı cache için idealdir.
• volatile-lru: Sadece TTL tanımlı anahtarlar arasında LRU uygular.
• allkeys-lfu: En az sıklıkla erişilen anahtarı siler (Redis 4.0+).
• volatile-ttl: TTL'i en kısa olan anahtarı öncelikli siler.

Redis ile Message Queue Kullanımı

Redis, yalnızca bir cache çözümü değildir; aynı zamanda güçlü bir mesaj kuyruğu (message queue) altyapısı olarak da kullanılabilir. Mikro servis mimarilerinde servisler arası iletişim, iş kuyruğu yönetimi ve olay tabanlı sistemlerde Redis farklı yaklaşımlar sunar.

List Tabanlı Basit Kuyruk

Redis'in list veri yapısı, LPUSH/RPOP veya RPUSH/LPOP komutlarıyla basit bir FIFO kuyruğu oluşturmak için kullanılabilir. BRPOP komutu ise bloklayıcı bir okuma sağlar ve polling'e gerek kalmadan yeni mesajları bekler.

# Producer (Üretici)
import redis
import json
import time

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

def enqueue_task(queue_name, task_data):
    message = json.dumps({
        'id': str(time.time()),
        'data': task_data,
        'created_at': time.strftime('%Y-%m-%d %H:%M:%S')
    })
    r.lpush(queue_name, message)
    print(f"Görev kuyruğa eklendi: {task_data}")

# Görevleri kuyruğa ekle
enqueue_task("email_queue", {"to": "user@example.com", "subject": "Hoş geldiniz"})
enqueue_task("email_queue", {"to": "admin@example.com", "subject": "Rapor hazır"})

# Consumer (Tüketici)
import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

def process_tasks(queue_name):
    print(f"Kuyruk dinleniyor: {queue_name}")
    while True:
        # BRPOP: Bloklayıcı okuma, 0 = süresiz bekle
        result = r.brpop(queue_name, timeout=0)
        if result:
            _, message = result
            task = json.loads(message)
            print(f"İşleniyor: {task['data']}")
            handle_task(task)

def handle_task(task):
    # Gerçek iş mantığı burada
    print(f"E-posta gönderiliyor: {task['data']['to']}")

process_tasks("email_queue")

Pub/Sub ile Gerçek Zamanlı Mesajlaşma

Redis Pub/Sub, yayıncı-abone modeli ile gerçek zamanlı mesajlaşma sağlar. Bir kanala yayınlanan mesaj, o anda kanala abone olan tüm istemcilere anında iletilir. Ancak dikkat edilmesi gereken önemli bir nokta vardır: Pub/Sub'da mesaj kalıcılığı yoktur. Abone olmayan bir istemci, kaçırdığı mesajları alamaz.

# Publisher (Yayıncı)
import redis

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# Kanala mesaj yayınla
r.publish("notifications", '{"type": "order", "message": "Yeni sipariş #1234"}')
r.publish("notifications", '{"type": "alert", "message": "Stok azaldı: Ürün X"}')

# Subscriber (Abone)
import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
pubsub = r.pubsub()
pubsub.subscribe("notifications")

print("Kanal dinleniyor...")
for message in pubsub.listen():
    if message['type'] == 'message':
        data = json.loads(message['data'])
        print(f"[{data['type']}] {data['message']}")

Redis Streams ile Güvenilir Mesaj Kuyruğu

Redis 5.0 ile birlikte gelen Streams, Pub/Sub ve List tabanlı kuyrukların eksiklerini gideren güçlü bir veri yapısıdır. Apache Kafka'ya benzer şekilde mesaj kalıcılığı, consumer group desteği ve mesaj onaylama (acknowledgment) mekanizması sunar.

# Producer - Stream'e mesaj ekleme
import redis

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# XADD ile stream'e mesaj ekle
message_id = r.xadd("order_stream", {
    "order_id": "1234",
    "customer": "Ahmet",
    "amount": "299.90",
    "status": "pending"
})
print(f"Mesaj eklendi, ID: {message_id}")

# Consumer Group oluşturma ve mesaj tüketme
import redis

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# Consumer group oluştur (ilk çalıştırmada)
try:
    r.xgroup_create("order_stream", "order_processors", id="0", mkstream=True)
except redis.exceptions.ResponseError:
    pass  # Grup zaten var

# Mesajları tüket
while True:
    messages = r.xreadgroup(
        groupname="order_processors",
        consumername="worker-1",
        streams={"order_stream": ">"},
        count=10,
        block=5000  # 5 saniye bekle
    )

    for stream, entries in messages:
        for msg_id, data in entries:
            print(f"İşleniyor: {data}")
            # İş mantığı...

            # Mesajı onayla (acknowledge)
            r.xack("order_stream", "order_processors", msg_id)

Streams'in avantajları:

• Mesaj kalıcılığı: Tüketilen mesajlar silinmez, geçmişe dönük okunabilir.
• Consumer Groups: Birden fazla tüketici aynı stream'i paylaşarak yük dağılımı yapabilir.
• Acknowledgment: Mesaj işleme başarısızlığında yeniden deneme mekanizması kurulabilir.
• XPENDING ile izleme: Onaylanmamış mesajları takip edebilirsiniz.

Performans Optimizasyonu İçin İpuçları

Pipeline Kullanımı

Redis'e birden fazla komut gönderirken her komut için ayrı bir round-trip yerine pipeline kullanarak tüm komutları tek seferde gönderebilirsiniz. Bu, ağ gecikmesini önemli ölçüde azaltır.

# Pipeline olmadan: her komut ayrı round-trip
for i in range(1000):
    r.set(f"key:{i}", f"value:{i}")

# Pipeline ile: tek round-trip
pipe = r.pipeline()
for i in range(1000):
    pipe.set(f"key:{i}", f"value:{i}")
pipe.execute()  # Tüm komutlar tek seferde gönderilir

Bağlantı Havuzu (Connection Pool)

Her istek için yeni bir bağlantı açmak yerine bağlantı havuzu kullanarak mevcut bağlantıları yeniden kullanabilirsiniz.

import redis

pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=50,
    decode_responses=True
)

# Her istemci havuzdan bağlantı alır
r = redis.Redis(connection_pool=pool)

Veri Serializasyonu

Büyük veri yapılarını cache'lerken serializasyon formatı performansı doğrudan etkiler. JSON okunabilir ama yavaştır; MessagePack veya Protobuf gibi binary formatlar daha hızlı ve daha kompakt sonuçlar verir.

import msgpack

# MessagePack ile serializasyon
data = {"user_id": 42, "name": "Elif", "scores": [95, 87, 92]}
packed = msgpack.packb(data)
r.set("user:42", packed)

# Deserializasyon
raw = r.get("user:42")
user = msgpack.unpackb(raw)
print(user)

Monitoring ve Hata Yönetimi

Üretim ortamında Redis'in sağlığını ve performansını izlemek kritiktir. Redis'in sunduğu bazı önemli metrikler:

• used_memory: Redis'in kullandığı toplam bellek miktarı.
• connected_clients: Aktif istemci bağlantı sayısı.
• keyspace_hits / keyspace_misses: Cache hit oranını hesaplamak için kullanılır.
• evicted_keys: Bellek sınırı nedeniyle silinen anahtar sayısı.

# Redis INFO komutu ile metrikleri sorgulama
info = r.info()
hit_rate = info['keyspace_hits'] / (info['keyspace_hits'] + info['keyspace_misses']) * 100
print(f"Cache hit oranı: {hit_rate:.2f}%")
print(f"Kullanılan bellek: {info['used_memory_human']}")
print(f"Bağlı istemci sayısı: {info['connected_clients']}")

Cache ve mesaj kuyruğu kullanırken hata yönetimi de ihmal edilmemelidir. Redis bağlantısının kopması, zaman aşımı veya bellek taşması gibi durumlara karşı uygulamanızda uygun fallback mekanizmaları bulunmalıdır.

from redis.exceptions import ConnectionError, TimeoutError

def get_with_fallback(key, fallback_fn):
    try:
        cached = r.get(key)
        if cached:
            return json.loads(cached)
    except (ConnectionError, TimeoutError) as e:
        print(f"Redis hatası: {e}")

    # Redis erişilemezse doğrudan kaynağa git
    return fallback_fn()

Sonuç

Redis, hem yüksek performanslı bir cache katmanı hem de esnek bir mesaj kuyruğu altyapısı olarak modern uygulamaların temel yapı taşlarından biridir. Cache-Aside stratejisinden Redis Streams'e kadar geniş bir yelpazede çözümler sunar. Doğru strateji seçimi, uygun TTL yönetimi, pipeline kullanımı ve dikkatli monitoring ile Redis'ten maksimum verimi alabilirsiniz.

Projenizin ihtiyaçlarına göre basit bir list tabanlı kuyruktan başlayıp, gerektiğinde Streams'e geçiş yapabilirsiniz. Önemli olan, her yaklaşımın güçlü ve zayıf yönlerini anlayarak doğru araçları doğru yerde kullanmaktır.