我們在昨天已經實現了對過期資料的被動刪除：當存取資料時，如果發現資料已過期就刪除它。
實作主動刪除功能？Redis 的做法是會建立一個執行緒，並且每 100 毫秒會從記憶體中取出一些資料來檢查是否需要刪除。
service.Worker 的重構我們之前就使用類似的方法建立 AOF 同步的功能，以該功能為基礎，可以建構一個 service.(*Worker) 來管理這些 gorutine，這樣可以保持 main.(*App).Run() 的邏輯更加乾淨：
// file: ./internal/service/worker.go
package service
import (
"context"
"errors"
"time"
"golang.org/x/sync/errgroup"
"olivine/internal/data"
"olivine/internal/repo"
)
type Worker interface {
Start(context.Context) error
}
func NewWorker(cfg *data.Config, aof AOF, storage repo.Storage) Worker {
return &worker{
cfg: cfg,
aof: aof,
storage: storage,
}
}
type worker struct {
cfg *data.Config
storage repo.Storage
aof AOF
}
func (w *worker) Start(ctx context.Context) error {
g, ctx := errgroup.WithContext(ctx)
if w.cfg.AOFEnabled && w.cfg.AOFFsync == data.AOFFsyncEverySec {
g.Go(w.aofSyncer(ctx))
}
g.Go(w.storagePruner(ctx))
return g.Wait()
}
func (w *worker) aofSyncer(ctx context.Context) func() error {
return func() error {
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return nil
case <-ticker.C:
if err := w.aof.Sync(); err != nil {
return err
}
}
}
}
}
現在，只要在 main.(*App).Run() 中調用 service.(*Worker).Start() 即可：
// file: ./cmd/olivine/app.go
func (app *App) Run() error {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
slog.Info("restoring data from disk")
if err := app.server.RestoreFromDisk(); err != nil {
return err
}
g, ctx := errgroup.WithContext(ctx)
g.Go(func() error {
slog.Info("starting olivine workers")
if err := app.worker.Start(ctx); err != nil {
return err
}
return nil
})
g.Go(func() error {
slog.Info("starting olivine server")
if err := app.server.ListenAndServe(); err != nil && !errors.Is(err, server.ErrServerClosed) {
return err
}
return nil
})
errch := make(chan error, 1)
go func() { errch <- g.Wait() }()
select {
case <-ctx.Done():
slog.Info("closing olivine server")
shutdownCtx, cancel := context.WithTimeout(context.Background(), time.Second*5)
defer cancel()
if err := app.server.Shutdown(shutdownCtx); err != nil {
return err
}
return <-errch
case err := <-errch:
return err
}
}
repo.Storage.Prune()在 repo.Storage 介面中加入 Prune()，service.(*Worker) 將會反覆調用它來進行過期資料清理。
// file: ./internal/repo/storage.go
func (s *mapStorage) Prune(ctx context.Context) error {
for {
select {
case <-ctx.Done():
return ctx.Err()
default:
if stop := s.tryPrune(); stop {
return nil
}
}
}
}
func (s *mapStorage) tryPrune() bool {
s.mu.Lock()
defer s.mu.Unlock()
if len(s.storage) == 0 {
return true
}
const sampleSize = 10
sampled := 0
expired := 0
now := time.Now()
for k, v := range s.storage {
if sampled == sampleSize {
break
}
sampled++
if v.ExpiresAt() != nil && now.After(*v.ExpiresAt()) {
delete(s.storage, k)
expired++
}
}
return expired*4 < sampled
}
在 service.(*Worker) 中加入 storagePruner()：
// file: internal/service/worker.go
func (w *worker) Start(ctx context.Context) error {
g, ctx := errgroup.WithContext(ctx)
if w.cfg.AOFEnabled && w.cfg.AOFFsync == data.AOFFsyncEverySec {
g.Go(w.aofSyncer(ctx))
}
+ g.Go(w.storagePruner(ctx))
return g.Wait()
}
+func (w *worker) storagePruner(ctx context.Context) func() error {
+ return func() error {
+ ticker := time.NewTicker(time.Millisecond * 100)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-ctx.Done():
+ return nil
+ case <-ticker.C:
+ if err := w.storage.Prune(ctx); err != nil {
+ if ctx.Err() != nil && errors.Is(err, ctx.Err()) {
+ return nil
+ }
+ return err
+ }
+ }
+ }
+ }
+}
目前，我們直接把 Redis 指令序列化為 RESP 然後存進 AOF 中，但這會出現一些未預期的問題：
我們來重新檢視一下整個流程：
SET foo bar EX 10，它會存進 AOF 與記憶體中GET foo，取得 bar，因為這個資料還沒有過期GET foo，取得 (nil)，因為這個資料已經過期foo -> bar 也會一起被還原GET foo，取得 bar，因為對服務來說這個資料還沒有過期，但事實上已經在 00:00:05 之後就過期了為了解決這個問題，在 AOF 中儲存的資料就不能是 幾秒後 或 幾毫秒後 這種資料，而應該直接儲存 unixtimestamp。
Redis 的官方實作中，會把 SET foo bar EX 10 這種指令直接轉換成 SET foo bar PXAT {} 這種指令後才存入 AOF 中
resp.(*Command) 與 AOF我們已經知道，實際收到的 Redis 指令與放入 AOF 的指令並不相同，因此我們在 resp.(*Command) 中額外定義一個欄位：
type Command struct {
raw Value
+ aof Array
cmd BulkString
args []BulkString
}
aof 會是一個 raw 的深拷貝，當指令是 SET 指令並且存在過期時間的時候會把 aof 另外設為 PXAT 的形式：
// file: ./pkg/resp/cmd.go
func ReadCommand(rd *Reader) (*Command, error) {
v, err := rd.Read()
if err != nil {
return nil, err
}
arr, ok := v.(Array)
if !ok {
return nil, fmt.Errorf("%w: expected array, got %T(%+v)", ErrProtocol, v, v)
}
if arr.null || len(arr.data) == 0 {
return nil, fmt.Errorf("%w: empty array", ErrProtocol)
}
cmd, ok := arr.data[0].(BulkString)
if !ok {
return nil, fmt.Errorf("%w: command expected bulk string, got %T(%+v)", ErrProtocol, arr.data[0], arr.data[0])
}
args := make([]BulkString, 0, len(arr.data)-1)
for i := range arr.data[1:] {
arg, ok := arr.data[i+1].(BulkString)
if !ok {
return nil, fmt.Errorf("%w: argument [%d] expected bulk string, got %T(%+v)", ErrProtocol, i, arr.data[i+1], arr.data[i+1])
}
args = append(args, arg)
}
return &Command{
raw: v,
aof: arr.Clone(),
cmd: cmd,
args: args,
}, nil
}
並且定義 func (cmd *Command) UpdateAOF(i int, v Value) {}：
// file: ./pkg/resp/cmd.go
func (cmd *Command) UpdateAOF(i int, v Value) {
cmd.aof.data[i] = v
}
並且在 SET 指令中調用 resp.(*Command).UpdateAOF()：
// file: ./internal/service/cmd/set.go
if p.Exp != nil && !p.Exp.IsZero() {
cmd.UpdateAOF(expirationOptionIndex+1, resp.NewBulkString("PXAT"))
cmd.UpdateAOF(expirationOptionIndex+2, resp.NewBulkString(strconv.FormatInt(p.Exp.UnixMilli(), 10)))
}
在 GitHub 上撰寫 [Day 13]Key Expiration 的時候其實遇上了滿大的困難，因為這個功能看似簡單卻充滿了理論與實務的權衡，且牽涉到大量的電腦科學前置知識，如果要深入探討會讓文章變得非常冗長。
我曾經試著解釋為什麼 Redis 設計為主動模式與被動模式，甚至不惜加入 IO Thread 來做過期資料的主動回收，而不是使用像 priority queue 這樣的資料結構來維護；我也想過是不是應該介紹一下在 Redis 底層是如何實作 redisObject，並且利用像魔法一樣的位操作來記錄過期時間；我也想跟你解釋為什麼我後來在解析指令參數時採用了有限狀態機的方案，而非使用正則表達式。
然而，為了文章的精簡跟易讀性，我前前後後修改了許多次，但也挺遺憾沒能把這些都一五一十地說完。
iThome鐵人賽
看影片追技術