Golang分佈式應用之Redis示例詳解

正文

Redis作是一個高性能的內存數據庫,常被應用於分佈式系統中,除瞭作為分佈式緩存或簡單的內存數據庫還有一些特殊的應用場景,本文結合Golang來編寫對應的中間件。

本文所有代碼見github.com/qingwave/go…

分佈式鎖

單機系統中我們可以使用sync.Mutex來保護臨界資源,在分佈式系統中同樣有這樣的需求,當多個主機搶占同一個資源,需要加對應的“分佈式鎖”。

在Redis中我們可以通過setnx命令來實現

  • 如果key不存在可以設置對應的值,設置成功則加鎖成功,key不存在返回失敗
  • 釋放鎖可以通過del實現。

主要邏輯如下:

type RedisLock struct {
	client     *redis.Client
	key        string
	expiration time.Duration // 過期時間,防止宕機或者異常
}
func NewLock(client *redis.Client, key string, expiration time.Duration) *RedisLock {
	return &RedisLock{
		client:     client,
		key:        key,
		expiration: expiration,
	}
}
// 加鎖將成功會將調用者id保存到redis中
func (l *RedisLock) Lock(id string) (bool, error) {
	return l.client.SetNX(context.TODO(), l.key, id, l.expiration).Result()
}
const unLockScript = `
if (redis.call("get", KEYS[1]) == KEYS[2]) then
	redis.call("del", KEYS[1])
	return true
end
return false
`
// 解鎖通過lua腳本來保證原子性,隻能解鎖當前調用者加的鎖
func (l *RedisLock) UnLock(id string) error {
	_, err := l.client.Eval(context.TODO(), unLockScript, []string{l.key, id}).Result()
	if err != nil && err != redis.Nil {
		return err
	}
	return nil
}

需要加一個額外的超時時間來防止系統宕機或者異常請求造成的死鎖,通過超時時間為最大預估運行時間的2倍。

解鎖時通過lua腳本來保證原子性,調用者隻會解自己加的鎖。避免由於超時造成的混亂,例如:進程A在時間t1獲取瞭鎖,但由於執行緩慢,在時間t2鎖超時失效,進程B在t3獲取瞭鎖,這是如果進程A執行完去解鎖會取消進程B的鎖。

運行測試

func main() {
    client := redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "123456",
		DB:       0, // use default DB
	})
	lock := NewLock(client, "counter", 30*time.Second)
    counter := 0
	worker := func(i int) {
		for {
			id := fmt.Sprintf("worker%d", i)
			ok, err := lock.Lock(id)
			log.Printf("worker %d attempt to obtain lock, ok: %v, err: %v", i, ok, err)
			if !ok {
				time.Sleep(100 * time.Millisecond)
				continue
			}
			defer lock.UnLock(id)
			counter++
			log.Printf("worker %d, add counter %d", i, counter)
			break
		}
	}
	wg := sync.WaitGroup{}
	for i := 1; i <= 5; i++ {
		wg.Add(1)
		id := i
		go func() {
			defer wg.Done()
			worker(id)
		}()
	}
	wg.Wait()
}

運行結果,可以看到與sync.Mutex使用效果類似

2022/07/22 09:58:09 worker 5 attempt to obtain lock, ok: true, err: <nil>
2022/07/22 09:58:09 worker 5, add counter 1
2022/07/22 09:58:09 worker 4 attempt to obtain lock, ok: false, err: <nil>
2022/07/22 09:58:09 worker 1 attempt to obtain lock, ok: false, err: <nil>
2022/07/22 09:58:09 worker 2 attempt to obtain lock, ok: false, err: <nil>
2022/07/22 09:58:09 worker 3 attempt to obtain lock, ok: false, err: <nil>
2022/07/22 09:58:10 worker 3 attempt to obtain lock, ok: false, err: <nil>
2022/07/22 09:58:10 worker 1 attempt to obtain lock, ok: false, err: <nil>
2022/07/22 09:58:10 worker 2 attempt to obtain lock, ok: false, err: <nil>
2022/07/22 09:58:10 worker 4 attempt to obtain lock, ok: true, err: <nil>
2022/07/22 09:58:10 worker 4, add counter 2
2022/07/22 09:58:10 worker 1 attempt to obtain lock, ok: true, err: <nil>
2022/07/22 09:58:10 worker 1, add counter 3
2022/07/22 09:58:10 worker 3 attempt to obtain lock, ok: false, err: <nil>
2022/07/22 09:58:10 worker 2 attempt to obtain lock, ok: false, err: <nil>
2022/07/22 09:58:10 worker 2 attempt to obtain lock, ok: true, err: <nil>
2022/07/22 09:58:10 worker 2, add counter 4
2022/07/22 09:58:10 worker 3 attempt to obtain lock, ok: false, err: <nil>
2022/07/22 09:58:10 worker 3 attempt to obtain lock, ok: true, err: <nil>
2022/07/22 09:58:10 worker 3, add counter 5

特別註意的是,在分佈式Redis集群中,如果發生異常時(主節點宕機),可能會降低分佈式鎖的可用性,可以通過強一致性的組件etcd、ZooKeeper等實現。

分佈式過濾器

假設要開發一個爬蟲服務,爬取百萬級的網頁,怎麼判斷某一個網頁是否爬取過,除瞭借助數據庫和HashMap,我們可以借助佈隆過濾器來做。相比其他方式佈隆過濾器占用極低的空間,而且插入查詢時間非常快。

佈隆過濾器用來判斷某個元素是否在集合中,利用BitSet

  • 插入數據時將值進行多次Hash,將BitSet對應位置1
  • 查詢時同樣進行多次Hash對比所有位上是否為1,如是則存在。

佈隆過濾器有一定的誤判率,不適合精確查詢的場景。另外也不支持刪除元素。通常適用於URL去重、垃圾郵件過濾、防止緩存擊穿等場景中。

在Redis中,我們可以使用自帶的BitSet實現,同樣也借助lua腳本的原子性來避免多次查詢數據不一致。

const (
	// 插入數據,調用setbit設置對應位
	setScript = `
for _, offset in ipairs(ARGV) do
	redis.call("setbit", KEYS[1], offset, 1)
end
`
	// 查詢數據,如果所有位都為1返回true
	getScript = `
for _, offset in ipairs(ARGV) do
	if tonumber(redis.call("getbit", KEYS[1], offset)) == 0 then
		return false
	end
end
return true
`
)
type BloomFilter struct {
	client *redis.Client
	key    string // 存在redis中的key
	bits   uint // BitSet的大小
	maps   uint // Hash的次數
}
func NewBloomFilter(client *redis.Client, key string, bits, maps uint) *BloomFilter {
	client.Del(context.TODO(), key)
	if maps == 0 {
		maps = 14
	}
	return &BloomFilter{
		key:    key,
		client: client,
		bits:   bits,
		maps:   maps,
	}
}
// 進行多次Hash, 得到位置列表
func (f *BloomFilter) getLocations(data []byte) []uint {
	locations := make([]uint, f.maps)
	for i := 0; i < int(f.maps); i++ {
		val := murmur3.Sum64(append(data, byte(i)))
		locations[i] = uint(val) % f.bits
	}
	return locations
}
func (f *BloomFilter) Add(data []byte) error {
	args := getArgs(f.getLocations(data))
	_, err := f.client.Eval(context.TODO(), setScript, []string{f.key}, args).Result()
	if err != nil && err != redis.Nil {
		return err
	}
	return nil
}
func (f *BloomFilter) Exists(data []byte) (bool, error) {
	args := getArgs(f.getLocations(data))
	resp, err := f.client.Eval(context.TODO(), getScript, []string{f.key}, args).Result()
	if err != nil {
		if err == redis.Nil {
			return false, nil
		}
		return false, err
	}
	exists, ok := resp.(int64)
	if !ok {
		return false, nil
	}
	return exists == 1, nil
}
func getArgs(locations []uint) []string {
	args := make([]string, 0)
	for _, l := range locations {
		args = append(args, strconv.FormatUint(uint64(l), 10))
	}
	return args
}

運行測試

func main() {
	bf := NewBloomFilter(client,"bf-test", 2^16, 14)
	exists, err := bf.Exists([]byte("test1"))
	log.Printf("exist %t, err %v", exists, err)
	if err := bf.Add([]byte("test1")); err != nil {
		log.Printf("add err: %v", err)
	}
	exists, err = bf.Exists([]byte("test1"))
	log.Printf("exist %t, err %v", exists, err)
	exists, err = bf.Exists([]byte("test2"))
	log.Printf("exist %t, err %v", exists, err)
// output
// 2022/07/22 10:05:58 exist false, err <nil>
// 2022/07/22 10:05:58 exist true, err <nil>
// 2022/07/22 10:05:58 exist false, err <nil>
}

分佈式限流器

golang.org/x/time/rate包中提供瞭基於令牌桶的限流器,如果要實現分佈式環境的限流可以基於Redis Lua腳本實現。

令牌桶的主要原理如下:

  • 假設一個令牌桶容量為burst,每秒按照qps的速率往裡面放置令牌
  • 初始時放滿令牌,令牌溢出則直接丟棄,請求令牌時,如果桶中有足夠令牌則允許,否則拒絕
  • 當burst==qps時,嚴格按照qps限流;當burst>qps時,可以允許一定的突增流量

這裡主要參考瞭官方rate包的實現,將核心邏輯改為Lua實現。

--- 相關Key
--- limit rate key值,對應value為當前令牌數
local limit_key = KEYS[1]
--- 輸入參數
--[[
qps: 每秒請求數;
burst: 令牌桶容量;
now: 當前Timestamp;
cost: 請求令牌數;
max_wait: 最大等待時間
--]]
local qps = tonumber(ARGV[1])
local burst = tonumber(ARGV[2])
local now = ARGV[3]
local cost = tonumber(ARGV[4])
local max_wait = tonumber(ARGV[5])
--- 獲取redis中的令牌數
local tokens = redis.call("hget", limit_key, "token")
if not tokens then
	tokens = burst
end
--- 上次修改時間
local last_time = redis.call("hget", limit_key, "last_time")
if not last_time then
	last_time = 0
end
--- 最新等待時間
local last_event = redis.call("hget", limit_key, "last_event")
if not last_event then
	last_event = 0
end
--- 通過當前時間與上次修改時間的差值,qps計算出當前時間得令牌數
local delta = math.max(0, now-last_time)
local new_tokens = math.min(burst, delta * qps + tokens)
new_tokens = new_tokens - cost --- 最新令牌數,減少請求令牌
--- 如果最新令牌數小於0,計算需要等待的時間
local wait_period = 0
if new_tokens < 0 and qps > 0 then
	wait_period = wait_period - new_tokens / qps
end
wait_period = math.ceil(wait_period)
local time_act = now + wait_period --- 滿足等待間隔的時間戳
--- 允許請求有兩種情況
--- 當請求令牌數小於burst, 等待時間不超過最大等待時間,可以通過補充令牌滿足請求
--- qps為0時,隻要最新令牌數不小於0即可
local ok = (cost <= burst and wait_period <= max_wait and qps > 0) or (qps == 0 and new_tokens >= 0)
--- 設置對應值
if ok then
	redis.call("set", limit_key, new_tokens)
	redis.call("set", last_time_key, now)
	redis.call("set", last_event_key, time_act)
end
--- 返回列表,{是否允許, 等待時間}
return {ok, wait_period}

在Golang中的相關接口Allow、AllowN、Wait等都是通過調用reserveN實現

// 調用lua腳本
func (lim *RedisLimiter) reserveN(now time.Time, n int, maxFutureReserveSecond int) (*Reservation, error) {
	// ...
	res, err := lim.rdb.Eval(context.TODO(), reserveNScript, []string{lim.limitKey}, lim.qps, lim.burst, now.Unix(), n, maxFutureReserveSecond).Result()
	if err != nil && err != redis.Nil {
		return nil, err
	}
	//...
	return &Reservation{
		ok:        allow == 1,
		lim:       lim,
		tokens:    n,
		timeToAct: now.Add(time.Duration(wait) * time.Second),
	}, nil
}

運行測試

func main() {
	rdb := redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "123456",
		DB:       0, // use default DB
	})
	r, err := NewRedisLimiter(rdb, 1, 2, "testrate")
	if err != nil {
		log.Fatal(err)
	}
	r.Reset()
	for i := 0; i < 5; i++ {
		err := r.Wait(context.TODO())
		log.Printf("worker %d allowed: %v", i, err)
	}
}
// output
// 2022/07/22 12:50:31 worker 0 allowed: <nil>
// 2022/07/22 12:50:31 worker 1 allowed: <nil>
// 2022/07/22 12:50:32 worker 2 allowed: <nil>
// 2022/07/22 12:50:33 worker 3 allowed: <nil>
// 2022/07/22 12:50:34 worker 4 allowed: <nil>

前兩個請求在burst內,直接可以獲得,後面的請求按照qps的速率生成。

其他

除此之外,Redis還可以用作全局計數、去重(set)、發佈訂閱等場景。Redis官方也提供瞭一些通用模塊,通過加載這些模塊也可以實現過濾、限流等特性,參考modules。

參考

https://go-zero.dev/

https://github.com/qingwave/gocorex

以上就是Golang分佈式應用之Redis示例詳解的詳細內容,更多關於Go分佈式Redis的資料請關註WalkonNet其它相關文章!

推薦閱讀: