GoSuda

Meningkatkan Responsivitas dengan Cache Sisi Klien Redis

By snowmerak views ...

Apa itu Redis?

Saya rasa tidak banyak orang yang tidak mengenal Redis. Namun, untuk menyebutkannya secara singkat dengan beberapa karakteristik, dapat dirangkum sebagai berikut.

  • Operasi dieksekusi pada single thread, sehingga semua operasi memiliki sifat atomisitas.
  • Data disimpan dan diolah di In-Memory, sehingga semua operasi berlangsung cepat.
  • Redis dapat menyimpan WAL sesuai opsi, sehingga status terbaru dapat di-backup dan dipulihkan dengan cepat.
  • Redis mendukung berbagai tipe data seperti Set, Hash, Bit, List, sehingga memiliki produktivitas yang tinggi.
  • Redis memiliki komunitas yang besar, sehingga pengalaman, isu, dan solusi yang beragam dapat dibagikan.
  • Redis dikembangkan dan dioperasikan dalam waktu yang lama, sehingga memiliki stabilitas yang dapat diandalkan.

Langsung ke Pokok Bahasan

Bayangkan?

Bagaimana jika cache layanan Anda memenuhi dua kondisi berikut?

  1. Data yang sering diakses harus disajikan kepada pengguna dalam keadaan terbaru, namun pembaruannya tidak teratur sehingga pembaruan cache harus dilakukan secara sering
  2. Data tidak diperbarui, tetapi data cache yang sama sering diakses dan di-query

Kasus pertama dapat dicontohkan dengan peringkat popularitas real-time pada toko daring. Ketika peringkat popularitas real-time toko daring disimpan dalam sorted set, akan tidak efisien jika Redis dibaca setiap kali pengguna mengakses halaman utama.
Kasus kedua berkaitan dengan data kurs valuta asing. Meskipun data kurs ditetapkan sekitar setiap 10 menit, pembacaannya terjadi sangat sering. Terutama untuk won-dolar, won-yen, dan won-yuan, cache dibaca dengan sangat sering.
Dalam kasus-kasus seperti ini, akan lebih efisien jika server API memiliki cache lokal tersendiri, lalu ketika data berubah, Redis dibaca kembali untuk melakukan pembaruan.

Lalu, bagaimana cara mewujudkan perilaku ini dalam arsitektur database - Redis - server API?

Bagaimana dengan Redis PubSub?

Saat menggunakan cache, mari berlangganan ke channel yang dapat menerima informasi apakah terjadi pembaruan!

  • Maka perlu dibuat logika yang mengirimkan pesan saat terjadi pembaruan.
  • Karena ada operasi tambahan akibat PubSub, performa akan terpengaruh.

pubsub-write

pubsub-read

Bagaimana jika Redis yang mendeteksi perubahan?

Bagaimana jika kita menerima notifikasi perintah untuk key tertentu menggunakan Keyspace Notification?

  • Diperlukan usaha untuk menyimpan dan membagikan key dan perintah yang digunakan untuk pembaruan sebelumnya.
  • Misalnya, untuk suatu key, perintah pembaruannya adalah Set biasa, sementara untuk key lain perintah pembaruannya adalah LPush, RPush, SAdd, atau SRem, sehingga menjadi rumit.
  • Hal ini sangat meningkatkan kemungkinan terjadinya miskomunikasi dalam pengembangan dan kesalahan manusia dalam pengodean.

Bagaimana jika kita menerima notifikasi per perintah menggunakan Keyevent Notification?

  • Diperlukan langganan ke semua perintah yang digunakan untuk pembaruan. Perlu dilakukan penyaringan yang tepat terhadap key yang masuk.
  • Misalnya, di antara semua key yang masuk melalui Del, untuk sebagian key, klien tersebut kemungkinan besar tidak memiliki cache lokal.
  • Hal ini dapat menyebabkan pemborosan sumber daya yang tidak perlu.

Maka yang dibutuhkan adalah Invalidation Message!

Apa itu Invalidation Message?

Invalidation Messages adalah konsep yang disediakan sebagai bagian dari Server Assisted Client-Side Cache yang ditambahkan sejak Redis 6.0. Invalidation Message disampaikan melalui alur berikut.

  1. Misalkan ClientB sudah pernah membaca key tersebut sekali.
  2. ClientA menetapkan ulang key tersebut.
  3. Redis mendeteksi perubahan dan menerbitkan Invalidation Message ke ClientB untuk memberitahukan agar ClientB menghapus cache-nya.
  4. ClientB menerima pesan tersebut dan mengambil tindakan yang sesuai.

invalidation-message

Bagaimana cara menggunakannya?

Struktur dasar operasi

Klien yang terhubung ke Redis menjalankan CLIENT TRACKING ON REDIRECT <client-id> agar dapat menerima invalidation message. Lalu, klien yang harus menerima pesan tersebut berlangganan ke SUBSCRIBE __redis__:invalidate untuk menerima invalidation message.

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"

Implementasi! Implementasi! Implementasi!

Redigo + Ristretto

Jika hanya dijelaskan seperti itu, cara penggunaannya dalam kode sebenarnya menjadi kurang jelas. Oleh karena itu, kita akan terlebih dahulu membangunnya secara sederhana dengan redigo dan ristretto.

Pertama, instal kedua dependensi berikut.

  • 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,     // jumlah key untuk melacak frekuensi (10M).
22		MaxCost:     1 << 30, // biaya maksimum cache (1GB).
23		BufferItems: 64,      // jumlah key per buffer Get.
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}

Pertama, kita membuat RedisClient sederhana yang berisi ristretto dan redigo.

 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}

Kodenya agak rumit.

  • Untuk melakukan Tracking, satu koneksi tambahan dibuat. Ini dilakukan untuk mempertimbangkan agar PubSub tidak mengganggu operasi lainnya.
  • ID dari koneksi tambahan tersebut diambil, lalu Tracking pada koneksi yang digunakan untuk membaca data dialihkan (redirect) ke koneksi tersebut.
  • Setelah itu, invalidation message dilanggan.
  • Kode untuk memproses langganan cukup rumit. Karena redigo tidak mem-parsing invalidation message, respons harus diterima dan diproses sebelum di-parsing.
 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}

Method Get mencari di ristretto terlebih dahulu seperti berikut, dan jika tidak ditemukan, data diambil dari Redis.

 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}

Kode untuk pengujian adalah seperti di atas. Jika Anda mencobanya, Anda akan melihat bahwa nilai diperbarui setiap kali data di Redis berubah.

Namun, pendekatan ini terlalu rumit. Terlebih lagi, untuk melakukan skala pada cluster, perlu mengaktifkan Tracking pada semua master atau replika.

Rueidis

Selama menggunakan bahasa Go, kita memiliki rueidis yang paling modern dan berkembang. Mari kita tulis kode yang menggunakan server assisted client side cache dalam lingkungan Redis cluster dengan rueidis.

Pertama, instal dependensi berikut.

  • github.com/redis/rueidis

Kemudian, tulis kode untuk membaca data dari Redis.

 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}

Untuk menggunakan client side cache di rueidis, cukup panggil DoCache. Dengan demikian, data ditambahkan ke cache lokal dengan pengaturan seperti berapa lama data akan dipertahankan di cache lokal, dan jika DoCache dipanggil kembali, data akan diambil dari dalam cache lokal. Tentu saja, invalidation message juga ditangani dengan benar.

Mengapa tidak redis-go?

Sayangnya, redis-go tidak mendukung server assisted client side cache melalui API resmi. Selain itu, saat membuat PubSub, koneksi baru dibuat dan tidak ada API untuk mengakses koneksi tersebut secara langsung, sehingga client id pun tidak dapat diketahui. Oleh karena itu, kami menilai bahwa konfigurasi dengan redis-go tidak mungkin dilakukan, dan kami melewatkannya.

Keren Juga

Melalui struktur client side cache

  • Jika data dapat disiapkan sebelumnya, melalui struktur ini kueri dan lalu lintas ke Redis dapat diminimalkan sambil tetap menyediakan data terbaru setiap saat.
  • Dengan ini, dapat dibangun semacam struktur CQRS sehingga performa membaca dapat meningkat secara signifikan.

cqrs

Seberapa keren hasilnya?

Kebetulan di lapangan, struktur seperti ini sedang digunakan, sehingga kami mencari latensi sederhana untuk dua API. Mohon maklum karena hanya dapat dituliskan secara sangat abstrak.

  1. API pertama
    1. Saat pertama kali diakses: rata-rata 14,63 ms
    2. Saat diakses berikutnya: rata-rata 2,82 ms
    3. Selisih rata-rata: 10,98 ms
  2. API kedua
    1. Saat pertama kali diakses: rata-rata 14,05 ms
    2. Saat diakses berikutnya: rata-rata 1,60 ms
    3. Selisih rata-rata: 11,57 ms

Peningkatan latensi tambahan mencapai sekitar 82% pada yang terbesar!

Kami memperkirakan terdapat peningkatan berikut.

  • Penghapusan proses komunikasi jaringan antara klien dan Redis serta penghematan lalu lintas
  • Penurunan jumlah perintah baca yang harus dijalankan oleh Redis itu sendiri
    • Hal ini juga memberikan efek peningkatan performa penulisan.
  • Minimalisasi parsing protokol Redis
    • Parsing protokol Redis pun tidak tanpa biaya. Mengurangi hal ini merupakan peluang besar.

Namun, semua hal merupakan trade-off. Untuk ini, kami setidaknya mengorbankan dua hal berikut.

  • Diperlukan implementasi, operasi, dan pemeliharaan elemen pengelolaan client side cache
  • Peningkatan penggunaan CPU dan memori pada klien akibat hal tersebut

Kesimpulan

Secara pribadi, ini merupakan komponen arsitektur yang memuaskan, dan stres pada latensi serta server API juga sangat berkurang. Ke depannya, saya berpikir akan baik jika arsitektur dibangun dengan struktur seperti ini bila memungkinkan.