iT邦幫忙

2026 iThome 鐵人賽
DAY 14
0
Software Development

從零開始打造 Redis：以 Go 建立 Production Ready 應用程式 系列 第 23 篇

[Day 23] 過期處理（上）：資料結構與被動刪除

18th鐵人賽
39 瀏覽

  • 分享至 

  • xImage
    •  

目前，我們用下列的結構來儲存資料：

// file: ./inernal/repo/storage.go
type mapStorage struct {
	storage map[string]string
	mu      sync.RWMutex
}

顯然，目前的簡單資料結構並不能處理資料過期這個功能。

Redis 的資料過期機制

Redis keys are expired in two ways: a passive way and active way.
Redis 有兩種資料過期機制：主動與被動

A key is passively expired when a client tries to access it and the key is timed out.
在被動模式中，過期的資料如果再次被存取時，Redis 會直接刪除它。

However, this is not enough as there are expired keys that will never be accessed again. These keys should be expired anyway, so periodically, Redis tests a few keys at random amongst the set of keys with an expiration. All the keys that are already expired are deleted from the keyspace.
然而，對於永遠不再被存取的資料，這樣的過期機制顯然是不夠的。因為這些資料不再被存取，但它們仍應該被刪除，因此 Redis 會隨機取出一些存在過期時間的資料，如果發現它們已經過期便會刪除它們。

Ref

因此，首先我們需要把 string 改成一個支援 expiration 的值。

object.Object

Object 物件 並不是指 Object-oriented programming 物件導向程式設計 中的「物件」，我們可以想像成它只是一個值的抽象，在 Go 中就是一個自定義的結構體。

// file: ./internal/repo/object/object.go
package object

import "time"

type Object interface {
	Key() string
	ExpiresAt() *time.Time
}

type base struct {
	key       string
	expiresAt *time.Time
}

func (obj base) Key() string {
	return obj.key
}

func (obj base) ExpiresAt() *time.Time {
	return obj.expiresAt
}

然後，定義「字串」是一個可被 object.Object 儲存的一種型態：

// file: ./internal/repo/object/string.go
package object

import "time"

type String struct {
	base

	val string
}

func NewString(k, v string, expiresAt *time.Time) *String {
	return &String{
		val:  v,
		base: base{key: k, expiresAt: expiresAt},
	}
}

func (str *String) String() string {
	return str.val
}

接著，我們可以更新 repo.(*mapStorage) 中的 string：

// file: internal/repo/storage.go
func NewStorage() Storage {
	s := mapStorage{}
-	s.storage = make(map[string]string)
+	s.storage = make(map[string]object.Object)

	return &s
}

type mapStorage struct {
-	storage map[string]string
+	storage map[string]object.Object
	mu      sync.Mutex
}

func (s *mapStorage) Set(_ context.Context, obj object.Object) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.storage[obj.Key()] = obj
	return nil
}

func (s *mapStorage) Get(_ context.Context, k string) (v object.Object, err error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	var ok bool
	if v, ok = s.storage[k]; !ok {
		return nil, fmt.Errorf("%w: miss", ErrNotFound)
	}
	if v.ExpiresAt() != nil && time.Now().After(*v.ExpiresAt()) {
		delete(s.storage, k) // remove key from storage when expired
		return nil, fmt.Errorf("%w: expired", ErrNotFound)
	}

	return
}

可以觀察到，當 Get 被呼叫時，它會先檢查資料是否過期，如果已經過期則會直接用 delete(s.storage, k) 刪除資料，順手就把被動刪除模式做完了。

SET 指令的格式

事實上，SET 這個 Redis 指令可以直接設定資料的過期時間。

註：其實以前 Redis 有很多像是 SETEX, SETNX, ... 之類的指令，但它們後來都被宣告為 deprecated 並且建議直接使用 SET 指令，因此在這個專案中我就不一一實作這些已被棄用的指令。

  • EX {sec}：資料將會在 sec 秒後過期
  • PX {msec}：資料將會在 msec 毫秒後過期
  • EXAT {usec} 資料將會在 usec 這個 unixtime 過期（以秒為單位）
  • PXAT {musec} 資料將會在 musec 這個 unixtime 過期（以毫秒為單位）

以下是完整的 SET 指令參數：

SET key value [NX | XX | IFEQ ifeq-value | IFNE ifne-value |
  IFDEQ ifdeq-digest | IFDNE ifdne-digest] [GET] [EX seconds |
  PX milliseconds | EXAT unix-time-seconds |
  PXAT unix-time-milliseconds | KEEPTTL]

我們可以建立一個 cmd.parse() 來處理這個指令參數，並且回傳 (parsedSet, error)：

const (
	nx    = iota + 1 // set value only not exists
	xx               // set value only exists
	ifeq             // set only value == cond.Val
	ifne             // set only value != cond.Val
	ifdeq            // set only XXH3(value) == cond.Val
	ifdne            // set only XXH3(value) != cond.Val
)

type parseSetCond struct {
	Typ int
	Val string
}

type parsedSet struct {
	K    string
	V    string
	Cond parseSetCond
	Get  bool
	Exp  *time.Time
}

func (c *Set) parse(cmd *resp.Command) (p parsedSet, err error) {
	const (
		awaitingKey = iota
		awaitingValue
		awaitingOption
		awaitingConditionValue
		awaitingPostConditionOption
		awaitingPostGetOption
		awaitingExpirationValue
		done
		invalid
	)

	state := awaitingKey
	var conditionType int
	var expirationOption string
	var expirationOptionIndex int

	for i, arg := range cmd.Args() {
		switch state {
		case awaitingKey:
			p.K = arg.String()
			state = awaitingValue
		case awaitingValue:
			p.V = arg.String()
			state = awaitingOption
		case awaitingOption:
			switch strings.ToUpper(arg.String()) {
			case "NX":
				p.Cond.Typ = nx
				state = awaitingPostConditionOption
			case "XX":
				p.Cond.Typ = xx
				state = awaitingPostConditionOption
			case "IFEQ":
				conditionType = ifeq
				state = awaitingConditionValue
			case "IFNE":
				conditionType = ifne
				state = awaitingConditionValue
			case "IFDEQ":
				conditionType = ifdeq
				state = awaitingConditionValue
			case "IFDNE":
				conditionType = ifdne
				state = awaitingConditionValue
			case "GET":
				p.Get = true
				state = awaitingPostGetOption
			case "EX", "PX", "EXAT", "PXAT":
				expirationOption = strings.ToUpper(arg.String())
				expirationOptionIndex = i
				state = awaitingExpirationValue
			case "KEEPTTL":
				p.Exp = new(time.Time)
				state = done
			default:
				state = invalid
			}
		case awaitingConditionValue:
			p.Cond = parseSetCond{Typ: conditionType, Val: arg.String()}
			state = awaitingPostConditionOption
		case awaitingPostConditionOption:
			switch strings.ToUpper(arg.String()) {
			case "GET":
				p.Get = true
				state = awaitingPostGetOption
			case "EX", "PX", "EXAT", "PXAT":
				expirationOption = strings.ToUpper(arg.String())
				expirationOptionIndex = i
				state = awaitingExpirationValue
			case "KEEPTTL":
				p.Exp = new(time.Time)
				state = done
			default:
				state = invalid
			}
		case awaitingPostGetOption:
			switch strings.ToUpper(arg.String()) {
			case "EX", "PX", "EXAT", "PXAT":
				expirationOption = strings.ToUpper(arg.String())
				expirationOptionIndex = i
				state = awaitingExpirationValue
			case "KEEPTTL":
				p.Exp = new(time.Time)
				state = done
			default:
				state = invalid
			}
		case awaitingExpirationValue:
			p.Exp, err = parseExpiration(expirationOption, arg.String())
			if err != nil {
				return parsedSet{}, fmt.Errorf("%w: %w", ErrSyntax, err)
			}
			state = done
		case done:
			state = invalid
		}
	}

	if state != awaitingOption && state != awaitingPostConditionOption && state != awaitingPostGetOption && state != done {
		return parsedSet{}, fmt.Errorf("%w: invalid SET syntax", ErrSyntax)
	}

	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)))
	}

	return p, nil
}

func parseExpiration(option, value string) (*time.Time, error) {
	switch option {
	case "EX", "PX":
		unit := "s"
		if option == "PX" {
			unit = "ms"
		}

		duration, err := parseDuration(value, unit)
		if err != nil {
			return nil, err
		}

		return new(time.Now().Add(duration)), nil
	case "EXAT", "PXAT":
		timestamp, err := strconv.Atoi(value)
		if err != nil {
			return nil, err
		}
		if timestamp <= 0 {
			return nil, errors.New("invalid expire time")
		}
		if option == "EXAT" {
			return new(time.Unix(int64(timestamp), 0)), nil
		}

		return new(time.UnixMilli(int64(timestamp))), nil
	}

	return nil, fmt.Errorf("unknown expiration option %q", option)
}

func parseDuration(s, unit string) (time.Duration, error) {
	if _, err := strconv.Atoi(s); err != nil {
		return 0, fmt.Errorf("invalid expire time: %w", err)
	}

	var sb strings.Builder
	sb.WriteString(s)
	sb.WriteString(unit)

	if d, err := time.ParseDuration(sb.String()); err != nil {
		return d, err
	} else if d <= 0 {
		return d, errors.New("invalid expire time")
	} else {
		return d, nil
	}
}

註：在這個解析器中，我使用了 Finiate State Machine 有限狀態機，它會讓程式碼變得非常長（因為要處理每個狀態），不過它比較易於理解狀態之間的切換。

repo.SetParam 介面

整理一下目前的流程：在 cmd.(Set) 指令使用 cmd.parse()，並且回傳 cmd.(*parsedSet) 作為結果放入 object.Object ，然後將其以 repo.Storage.Set() 存入記憶體。

此時，我們可以定義一個 repo.SetParam 介面用來處理從 cmd.(*parsedSet) 到 repo.Storage.Set() 的這段流程：

// file: ./internal/repo/storage.go
type SetParam interface {
	Obj() object.Object
}

並且讓 cmd.(*parsedSet) 實作這個 repo.SetParam 介面：

type parsedSet struct {
	K    string
	V    string
	Cond parseSetCond
	Get  bool
	Exp  *time.Time
}

var _ repo.SetParam = parsedSet{}

func (parsed parsedSet) Obj() object.Object {
	return object.NewString(parsed.K, parsed.V, parsed.Exp)
}

註：var _ repo.SetParam = parsedSet{} 這是一個介面實作常用的 pattern，它可以讓編譯器檢查 cmd.(*parsedSet) 是否是 repo.SetParam 的實作。

然後我們只要對 repo.Storage.Set() 稍加修改，就可以使用 repo.SetParam：

// file: ./internal/repo/storage.go
func (s *mapStorage) Set(_ context.Context, param SetParam) error { 
	s.mu.Lock() 
	defer s.mu.Unlock() 

	s.storage[param.Obj().Key()] = param.Obj() 
	return nil 
}

此處定義了 repo.SetParam 來處理 cmd.(*parsedSet) 的理由是為了讓 repo.Storage.Set() 可以不止接受 object.(*String)，同時也保留類似於 object.(*List) 這樣的可能性。


上一篇
[Day 22] 依賴注入
下一篇
[Day 24] 過期處理（下）：主動刪除與 AOF 處理
系列文
從零開始打造 Redis：以 Go 建立 Production Ready 應用程式 共 29 篇
  1. 25
    [Day 25] 當 SET 需要當前資料
  2. 26
    [Day 26] 用 pprof 進行性能剖析
  3. 27
    [Day 27] 對互斥鎖的性能剖析
  4. 28
    [Day 28] 對記憶體性能進行剖析（上）
  5. 29
    [Day 29] 對記憶體進行性能剖析（下）
完整目錄
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言
iThome鐵人賽
參賽組數
391 組
團體組數
15 組
累計文章數
4014 篇
最後報名日
9/15
iT+看影片追技術 看影片追技術 看更多
熱門tag
15th鐵人賽 16th鐵人賽 13th鐵人賽 14th鐵人賽 17th鐵人賽 12th鐵人賽 11th鐵人賽 鐵人賽 2019鐵人賽 javascript 2018鐵人賽 python 18th鐵人賽 2017鐵人賽 windows php c# linux windows server css
熱門問題
熱門回答
熱門文章
IT邦幫忙