Redis İstemci Tarafı Önbelleği ile Tepki Verme Hızını Artırma

Redis Nedir?

Redis'i tanımayan pek kimse olmadığını düşünüyorum. Yine de birkaç özelliğine kısaca değinip geçmek gerekirse, aşağıdaki şekilde özetlenebilir.

• Tek iş parçacığı (single thread) üzerinde işlemler gerçekleştirildiği için, tüm işlemler atomikliğe sahiptir.
• Veriler In-Memory olarak saklanır ve üzerinde işlem yapılır, bu nedenle tüm işlemler hızlıdır.
• Redis, seçeneklere bağlı olarak WAL kaydedebilir; bu sayede en güncel durumu hızlıca yedekleyebilir ve geri yükleyebilir.
• Set, Hash, Bit, List gibi çeşitli veri tiplerini desteklediği için yüksek üretkenlik sağlar.
• Geniş bir topluluğa sahiptir; bu sayede çeşitli deneyimleri, sorunları ve çözümleri paylaşabilirsiniz.
• Uzun süredir geliştirilip işletilmekte olduğu için güvenilir bir kararlılığa sahiptir.

Şimdi Asıl Konuya Gelelim

Bir Düşünün?

Hizmetinizdeki önbelleğin (cache) aşağıdaki iki koşula uyduğunu varsayalım.

1. Sık sorgulanan verilerin güncel durumunu kullanıcıya sunmanız gerekiyor, ancak güncellemeler düzensiz olduğu için önbelleği sık sık güncellemeniz gerektiğinde
2. Önbellek güncellenmiyor, ancak aynı önbellek verisine sık sık erişip sorgulamanız gerektiğinde

İlk durumun örneği olarak bir alışveriş sitesinin gerçek zamanlı popülerlik sıralamasını ele alabiliriz. Alışveriş sitesinin gerçek zamanlı popülerlik sıralamasını sorted set olarak sakladığımızda, kullanıcı ana sayfaya her eriştiğinde Redis'ten okumak verimsizdir. İkinci durumda ise döviz kuru verileri söz konusudur. Kurlar yaklaşık 10 dakikalık periyotlarla ilan edilse de gerçek sorgular çok sık gerçekleşir. Özellikle Won-Dolar, Won-Yen ve Won-Yuan için önbellek çok sık sorgulanır. Bu tür durumlarda API sunucusunun yerel olarak ayrı bir önbelleğe sahip olması ve veriler değiştiğinde Redis'i yeniden sorgulayarak güncellemesi daha verimli bir işleyiş olacaktır.

Peki, veritabanı - Redis - API sunucusu yapısında bu işleyişi nasıl uygulayabiliriz??

Redis PubSub ile Olmaz mı?

Önbelleği kullanırken, güncelleme durumunu alabileceğimiz bir kanala abone olalım!

• Bu durumda, güncelleme anında bir mesaj gönderen bir mantık oluşturmamız gerekir.
• PubSub'ın ek işlemleri eklenmesi performansı etkiler.

Peki Ya Redis Değişikliği Algılarsa?

Keyspace Notification kullanarak ilgili anahtar için komut bildirimleri alırsak?

• Güncellemede kullanılan anahtarları ve komutları önceden kaydetmek ve paylaşmak zahmetlidir.
• Örneğin, bazı anahtarlar için basit Set güncelleme komutuyken, bazıları için LPush, ya da RPush veya SAdd ve SRem güncelleme komutları olabilir; bu da işleri karmaşıklaştırır.
• Bu durum, geliştirme sürecinde iletişim hatalarına ve kodlamada insan hatalarına yol açma olasılığını önemli ölçüde artırır.

Keyevent Notification kullanarak komut bazında bildirim alırsak?

• Güncellemede kullanılan tüm komutlara abone olmak gerekir. Gelen anahtarlar için de uygun filtreleme gerekir.
• Örneğin, Del ile gelen tüm anahtarların bir kısmı için ilgili istemcinin yerel önbelleği olmayabilir.
• Bu da gereksiz kaynak israfına yol açabilir.

O Zaman İhtiyacımız Olan Şey Invalidation Message!

Invalidation Message Nedir?

Invalidation Messages, Redis 6.0 ile eklenen Server Assisted Client-Side Cache kapsamında sunulan bir kavramdır. Invalidation Message şu akışla iletilir.

1. ClientB'nin bir key'i daha önce bir kez okuduğunu varsayalım.
2. ClientA ilgili key'i yeniden ayarlar.
3. Redis değişikliği algılar ve ClientB'ye bir Invalidation Message yayınlayarak önbelleği temizlemesini bildirir.
4. ClientB bu mesajı alır ve uygun bir işlem gerçekleştirir.

Nasıl Kullanılır?

Temel Çalışma Yapısı

Redis'e bağlı istemci CLIENT TRACKING ON REDIRECT <client-id> komutunu çalıştırarak invalidation message almasını sağlar. Ardından mesajları alması gereken istemci SUBSCRIBE __redis__:invalidate ile invalidation message'a abone olur.

Default Tracking

1# client 1
2> SET a 100

1# client 3
2> CLIENT ID
312
4> SUBSCRIBE __redis__:invalidate
51) "subscribe"
62) "__redis__:invalidate"
73) (integer) 1

1# client 2
2> CLIENT TRACKING ON REDIRECT 12
3> GET a # tracking

1# client 1
2> SET a 200

1# client 3
21) "message"
32) "__redis__:invalidate"
43) 1) "a"

Broadcasting Tracking

1# client 3
2> CLIENT ID
312
4> SUBSCRIBE __redis__:invalidate
51) "subscribe"
62) "__redis__:invalidate"
73) (integer) 1

1# client 2
2CLIENT TRACKING ON BCAST PREFIX cache: REDIRECT 12

1# client 1
2> SET cache:name "Alice"
3> SET cache:age 26

1# client 3
21) "message"
32) "__redis__:invalidate"
43) 1) "cache:name"
51) "message"
62) "__redis__:invalidate"
73) 1) "cache:age"

Uygulama! Uygulama! Uygulama!

Redigo + Ristretto

Bunu yalnızca bu şekilde anlatmak, kodda gerçekten nasıl kullanılacağı konusunda belirsizlik bırakır. Bu nedenle önce redigo ve ristretto ile basit bir yapı oluşturalım.

Öncelikle iki bağımlılığı kuruyoruz.

• github.com/gomodule/redigo
• github.com/dgraph-io/ristretto

 1package main
 2
 3import (
 4	"context"
 5	"fmt"
 6	"log/slog"
 7	"time"
 8
 9	"github.com/dgraph-io/ristretto"
10	"github.com/gomodule/redigo/redis"
11)
12
13type RedisClient struct {
14	conn  redis.Conn
15	cache *ristretto.Cache[string, any]
16	addr  string
17}
18
19func NewRedisClient(addr string) (*RedisClient, error) {
20	cache, err := ristretto.NewCache(&ristretto.Config[string, any]{
21		NumCounters: 1e7,     // number of keys to track frequency of (10M).
22		MaxCost:     1 << 30, // maximum cost of cache (1GB).
23		BufferItems: 64,      // number of keys per Get buffer.
24	})
25	if err != nil {
26		return nil, fmt.Errorf("failed to generate cache: %w", err)
27	}
28
29	conn, err := redis.Dial("tcp", addr)
30	if err != nil {
31		return nil, fmt.Errorf("failed to connect to redis: %w", err)
32	}
33
34	return &RedisClient{
35		conn:  conn,
36		cache: cache,
37		addr:  addr,
38	}, nil
39}
40
41func (r *RedisClient) Close() error {
42	err := r.conn.Close()
43	if err != nil {
44		return fmt.Errorf("failed to close redis connection: %w", err)
45	}
46
47	return nil
48}

Önce ristretto ve redigo'yu içeren bir RedisClient oluşturuyoruz.

 1func (r *RedisClient) Tracking(ctx context.Context) error {
 2	psc, err := redis.Dial("tcp", r.addr)
 3	if err != nil {
 4		return fmt.Errorf("failed to connect to redis: %w", err)
 5	}
 6
 7	clientId, err := redis.Int64(psc.Do("CLIENT", "ID"))
 8	if err != nil {
 9		return fmt.Errorf("failed to get client id: %w", err)
10	}
11	slog.Info("client id", "id", clientId)
12
13	subscriptionResult, err := redis.String(r.conn.Do("CLIENT", "TRACKING", "ON", "REDIRECT", clientId))
14	if err != nil {
15		return fmt.Errorf("failed to enable tracking: %w", err)
16	}
17	slog.Info("subscription result", "result", subscriptionResult)
18
19	if err := psc.Send("SUBSCRIBE", "__redis__:invalidate"); err != nil {
20		return fmt.Errorf("failed to subscribe: %w", err)
21	}
22	psc.Flush()
23
24	for {
25		msg, err := psc.Receive()
26		if err != nil {
27			return fmt.Errorf("failed to receive message: %w", err)
28		}
29
30		switch msg := msg.(type) {
31		case redis.Message:
32			slog.Info("received message", "channel", msg.Channel, "data", msg.Data)
33			key := string(msg.Data)
34			r.cache.Del(key)
35		case redis.Subscription:
36			slog.Info("subscription", "kind", msg.Kind, "channel", msg.Channel, "count", msg.Count)
37		case error:
38			return fmt.Errorf("error: %w", msg)
39		case []interface{}:
40			if len(msg) != 3 || string(msg[0].([]byte)) != "message" || string(msg[1].([]byte)) != "__redis__:invalidate" {
41				slog.Warn("unexpected message", "message", msg)
42				continue
43			}
44
45			contents := msg[2].([]interface{})
46			keys := make([]string, len(contents))
47			for i, key := range contents {
48				keys[i] = string(key.([]byte))
49				r.cache.Del(keys[i])
50			}
51			slog.Info("received invalidation message", "keys", keys)
52		default:
53			slog.Warn("unexpected message", "type", fmt.Sprintf("%T", msg))
54		}
55	}
56}

Kod biraz karmaşık.

• Tracking için bir bağlantı daha açıyoruz. Bu, PubSub'ın diğer işlemleri engellemesi ihtimalini dikkate alan bir önlemdir.
• Eklenen bağlantının kimliğini sorgulayıp, verileri sorgulayacak bağlantıda Tracking'i bu bağlantıya yönlendiriyoruz (REDIRECT).
• Ardından invalidation message'a abone oluyoruz.
• Abonelik işlemini yöneten kod biraz karmaşıktır. redigo, geçersiz kılma mesajlarını ayrıştıramadığı için yanıtı ayrıştırmadan önce almak ve işlemek gerekir.

 1func (r *RedisClient) Get(key string) (any, error) {
 2	val, found := r.cache.Get(key)
 3	if found {
 4		switch v := val.(type) {
 5		case int64:
 6			slog.Info("cache hit", "key", key)
 7			return v, nil
 8		default:
 9			slog.Warn("unexpected type", "type", fmt.Sprintf("%T", v))
10		}
11	}
12	slog.Info("cache miss", "key", key)
13
14	val, err := redis.Int64(r.conn.Do("GET", key))
15	if err != nil {
16		return nil, fmt.Errorf("failed to get key: %w", err)
17	}
18
19	r.cache.SetWithTTL(key, val, 1, 10*time.Second)
20	return val, nil
21}

Get metodu, aşağıdaki gibi önce ristretto'yu sorgular, yoksa Redis'ten alır.

 1package main
 2
 3import (
 4	"context"
 5	"log/slog"
 6	"os"
 7	"os/signal"
 8	"time"
 9)
10
11func main() {
12	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
13	defer cancel()
14
15	client, err := NewRedisClient("localhost:6379")
16	if err != nil {
17		panic(err)
18	}
19	defer client.Close()
20
21	go func() {
22		if err := client.Tracking(ctx); err != nil {
23			slog.Error("failed to track invalidation message", "error", err)
24		}
25	}()
26
27	ticker := time.NewTicker(1 * time.Second)
28	defer ticker.Stop()
29	done := ctx.Done()
30
31	for {
32		select {
33		case <-done:
34			slog.Info("shutting down")
35			return
36		case <-ticker.C:
37			v, err := client.Get("key")
38			if err != nil {
39				slog.Error("failed to get key", "error", err)
40				return
41			}
42			slog.Info("got key", "value", v)
43		}
44	}
45}

Test için kullanılacak kod yukarıdaki gibidir. Bir kez deneyin; Redis'teki veri her güncellendiğinde değerin yeniden alındığını görebilirsiniz.

Ancak bu yaklaşım çok karmaşıktır. Üstelik kümeye (cluster) ölçeklenebilmek için zorunlu olarak tüm master veya replica'larda Tracking'i etkinleştirmek gerekir.

Rueidis

Go dilini kullandığımız sürece elimizde en modern ve gelişmiş olan rueidis bulunuyor. rueidis ile Redis cluster ortamında server assisted client side cache kullanan bir kod yazalım.

Öncelikle bağımlılığı kuruyoruz.

• github.com/redis/rueidis

Ardından Redis'ten veri sorgulayan kodu yazıyoruz.

 1package main
 2
 3import (
 4	"context"
 5	"log/slog"
 6	"os"
 7	"os/signal"
 8	"time"
 9
10	"github.com/redis/rueidis"
11)
12
13func main() {
14	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
15	defer cancel()
16
17	client, err := rueidis.NewClient(rueidis.ClientOption{
18		InitAddress: []string{"localhost:6379"},
19	})
20	if err != nil {
21		panic(err)
22	}
23
24	ticker := time.NewTicker(1 * time.Second)
25	defer ticker.Stop()
26	done := ctx.Done()
27
28	for {
29		select {
30		case <-done:
31			slog.Info("shutting down")
32			return
33		case <-ticker.C:
34			const key = "key"
35			resp := client.DoCache(ctx, client.B().Get().Key(key).Cache(), 10*time.Second)
36			if resp.Error() != nil {
37				slog.Error("failed to get key", "error", resp.Error())
38				continue
39			}
40			i, err := resp.AsInt64()
41			if err != nil {
42				slog.Error("failed to convert response to int64", "error", err)
43				continue
44			}
45			switch resp.IsCacheHit() {
46			case true:
47				slog.Info("cache hit", "key", key)
48			case false:
49				slog.Info("missed key", "key", key)
50			}
51			slog.Info("got key", "value", i)
52		}
53	}
54}

rueidis'te client side cache kullanmak için yalnızca DoCache çağırmak yeterlidir. Yerel önbellekte verinin ne kadar tutulacağı gibi ayarlar, yerel önbelleğe eklenirken belirtilir; aynı şekilde DoCache tekrar çağrıldığında veri yerel önbellekten alınır. Tabii ki invalidation message'ları da düzgün şekilde işler.

Neden redis-go Değil?

redis-go, maalesef resmi API ile server assisted client side cache desteği sunmuyor. Üstelik PubSub oluşturulurken yeni bir bağlantı açılıyor ve bu bağlantıya doğrudan erişen bir API bulunmadığı için client id'yi de öğrenmek mümkün değil. Bu nedenle redis-go ile yapılandırmanın mümkün olmadığına karar verdik ve bu seçeneği geçtik.

Ne Kadar Şık

Client Side Cache Yapısı Sayesinde

• Önceden hazırlanabilecek veriler için bu yapı sayesinde Redis'e yönelik sorgu ve trafik en aza indirilebilir ve her zaman güncel veri sunulabilir.
• Bu sayede bir tür CQRS yapısı oluşturularak okuma performansı önemli ölçüde artırılabilir.

Ne Kadar Daha Şık Hale Geldi?

Gerçekte sahada tam da bu yapıyla kullanım olduğu için iki API için kısa bir gecikme (latency) süresi incelemesi yaptık. Yalnızca çok soyut bir biçimde paylaşabildiğimiz için anlayışınızı rica ederiz.

1. Birinci API
   1. İlk sorgulamada: ortalama 14.63ms
   2. Sonraki sorgulamalarda: ortalama 2.82ms
   3. Ortalama fark: 10.98ms
2. İkinci API
   1. İlk sorgulamada: ortalama 14.05ms
   2. Sonraki sorgulamalarda: ortalama 1.60ms
   3. Ortalama fark: 11.57ms

En fazla yaklaşık %82 oranında ek gecikme iyileşmesi sağlandı!

Muhtemelen aşağıdaki iyileştirmeler gerçekleşmiştir:

• İstemci ile Redis arasındaki ağ iletişimi sürecinin atlanması ve trafik tasarrufu
• Redis'in kendisinin gerçekleştirmesi gereken okuma komutu sayısının azalması
  • Bu, yazma performansını da artırma etkisine sahiptir.
• Redis protokolünün ayrıştırılmasının (parsing) en aza indirilmesi
  • Redis protokolünü ayrıştırmanın maliyeti de sıfır değildir. Bunu azaltabilmek büyük bir fırsattır.

Ancak her şey bir ödünleşimdir (trade-off). Bunun için en azından aşağıdaki iki noktadan fedakârlık ettik:

• Client side cache yönetim unsurlarının uygulanması, işletilmesi ve bakımına duyulan ihtiyaç
• Bunun sonucunda istemcinin CPU ve bellek kullanımında artış

Sonuç

Kişisel olarak tatmin edici bir mimari bileşen oldu; gecikme ve API sunucusu üzerindeki yük de oldukça düşük kaldı. Mümkün olduğunda bundan sonra da bu yapıyla mimari tasarlanmasının iyi olacağını düşünüyorum.