文章目录
- 流程图
- Redis锁的深入实现
- Backoff重试策略的深入探讨
- 结合Redis锁与Backoff策略的高级应用
- 具体实现
- 结论
在构建分布式系统时，确保数据的一致性和操作的原子性是至关重要的。Redis锁作为一种高效且广泛使用的分布式锁机制，能够帮助我们在多进程或分布式环境中同步访问共享资源。本文将深入探讨如何在Go语言中实现Redis锁，并结合Backoff重试策略来优化锁的获取过程，确保系统的健壮性和可靠性。
流程图
Redis锁的深入实现
在Go语言中，我们使用github.com/gomodule/redigo/redis包来操作Redis。Redis锁的实现依赖于Redis的SET命令，该命令支持设置键值对，并且可以带有过期时间（EX选项）和仅当键不存在时才设置（NX选项）。以下是一个更详细的Redis锁实现示例：
func SetWithContext(ctx context.Context, redisPool *redis.Pool, key string, expireSecond uint32) (bool, string, error) {
// ...省略部分代码...
conn, err := redisPool.GetContext(ctx)
if err != nil {
return false, "", err
}
defer conn.Close()
randVal := generateRandVal() // 生成随机值
_, err = conn.Do("SET", key, randVal, "NX", "EX", int(expireSecond))
if err != nil {
return false, "", err
}
return true, randVal, nil
}
在上述代码中，generateRandVal()函数用于生成一个唯一的随机值，这个值在释放锁时用来验证是否是锁的持有者。expireSecond参数确保了即使客户端崩溃或网络问题发生，锁也会在一定时间后自动释放，避免死锁。
释放锁时，我们使用Lua脚本来确保只有持有锁的客户端才能删除键：
func ReleaseWithContext(ctx context.Context, redisPool *redis.Pool, key string, randVal string) error {
// ...省略部分代码...
conn, err := redisPool.GetContext(ctx)
if err != nil {
return err
}
defer conn.Close()
script := `
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
`
_, err = conn.Do("EVAL", script, 1, key, randVal)
return err
}
Backoff重试策略的深入探讨
在分布式系统中，获取锁可能会因为网络延迟、高负载或其他原因而失败。Backoff重试策略通过在重试之间引入等待时间来减轻这些问题的影响。在提供的代码中，我们定义了多种Backoff策略，每种策略都有其特定的使用场景和优势。
例如，指数退避策略ExponentialBackoff的实现如下：
func (b *ExponentialBackoff) Next(retry int) (time.Duration, bool) {
// ...省略部分代码...
m := math.Min(r*b.t*math.Pow(b.f, float64(retry)), b.m)
if m >= b.m {
return 0, false
}
d := time.Duration(int64(m)) * time.Millisecond
return d, true
}
在这个策略中，重试间隔随重试次数的增加而指数级增长，但有一个最大值限制。这有助于在遇到连续失败时，逐步增加等待时间，避免立即重载系统。
结合Redis锁与Backoff策略的高级应用
将Redis锁与Backoff策略结合起来，可以创建一个健壮的锁获取机制。例如，我们可以定义一个MustSetRetry方法，该方法会不断尝试获取锁，直到成功为止：
func (r *RedisLock) MustSetRetry(ctx context.Context, key string) (string, error) {
op := func() (string, error) {
return r.MustSet(ctx, key)
}
notifyFunc := func(err error) {
// ...错误处理逻辑...
}
return mustSetRetryNotify(op, r.backoff, notifyFunc)
}
在这个方法中，mustSetRetryNotify函数负责执行重试逻辑，直到MustSet方法成功获取锁或达到最大重试次数。通过这种方式，我们能够确保即使在高竞争环境下，也能以一种可控和安全的方式获取锁。
具体实现
- backoff
package lock
import (
"math"
"math/rand"
"sync"
"time"
)
// BackoffFunc specifies the signature of a function that returns the
// time to wait before the next call to a resource. To stop retrying
// return false in the 2nd return value.
type BackoffFunc func(retry int) (time.Duration, bool)
// Backoff allows callers to implement their own Backoff strategy.
type Backoff interface {
// Next implements a BackoffFunc.
Next(retry int) (time.Duration, bool)
}
// -- ZeroBackoff --
// ZeroBackoff is a fixed backoff policy whose backoff time is always zero,
// meaning that the operation is retried immediately without waiting,
// indefinitely.
type ZeroBackoff struct{}
// Next implements BackoffFunc for ZeroBackoff.
func (b ZeroBackoff) Next(retry int) (time.Duration, bool) {
return 0, true
}
// -- StopBackoff --
// StopBackoff is a fixed backoff policy that always returns false for
// Next(), meaning that the operation should never be retried.
type StopBackoff struct{}
// Next implements BackoffFunc for StopBackoff.
func (b StopBackoff) Next(retry int) (time.Duration, bool) {
return 0, false
}
// -- ConstantBackoff --
// ConstantBackoff is a backoff policy that always returns the same delay.
type ConstantBackoff struct {
interval time.Duration
}
// NewConstantBackoff returns a new ConstantBackoff.
func NewConstantBackoff(interval time.Duration) *ConstantBackoff {
return &ConstantBackoff{interval: interval}
}
// Next implements BackoffFunc for ConstantBackoff.
func (b *ConstantBackoff) Next(retry int) (time.Duration, bool) {
return b.interval, true
}
// -- Exponential --
// ExponentialBackoff implements the simple exponential backoff described by
// Douglas Thain at http://dthain.blogspot.de/2009/02/exponential-backoff-in-distributed.html.
type ExponentialBackoff struct {
t float64 // initial timeout (in msec)
f float64 // exponential factor (e.g. 2)
m float64 // maximum timeout (in msec)
}
// NewExponentialBackoff returns a ExponentialBackoff backoff policy.
// Use initialTimeout to set the first/minimal interval
// and maxTimeout to set the maximum wait interval.
func NewExponentialBackoff(initialTimeout, maxTimeout time.Duration) *ExponentialBackoff {
return &ExponentialBackoff{
t: float64(int64(initialTimeout / time.Millisecond)),
f: 2.0,
m: float64(int64(maxTimeout / time.Millisecond)),
}
}
// Next implements BackoffFunc for ExponentialBackoff.
func (b *ExponentialBackoff) Next(retry int) (time.Duration, bool) {
r := 1.0 + rand.Float64() // random number in [1..2]
m := math.Min(r*b.t*math.Pow(b.f, float64(retry)), b.m)
if m >= b.m {
return 0, false
}
d := time.Duration(int64(m)) * time.Millisecond
return d, true
}
// -- Simple Backoff --
// SimpleBackoff takes a list of fixed values for backoff intervals.
// Each call to Next returns the next value from that fixed list.
// After each value is returned, subsequent calls to Next will only return
// the last element. The values are optionally "jittered" (off by default).
type SimpleBackoff struct {
sync.Mutex
ticks []int
jitter bool
}
// NewSimpleBackoff creates a SimpleBackoff algorithm with the specified
// list of fixed intervals in milliseconds.
func NewSimpleBackoff(ticks ...int) *SimpleBackoff {
return &SimpleBackoff{
ticks: ticks,
jitter: false,
}
}
// Jitter enables or disables jittering values.
func (b *SimpleBackoff) Jitter(flag bool) *SimpleBackoff {
b.Lock()
b.jitter = flag
b.Unlock()
return b
}
// jitter randomizes the interval to return a value of [0.5*millis .. 1.5*millis].
func jitter(millis int) int {
if millis <= 0 {
return 0
}
return millis/2 + rand.Intn(millis)
}
// Next implements BackoffFunc for SimpleBackoff.
func (b *SimpleBackoff) Next(retry int) (time.Duration, bool) {
b.Lock()
defer b.Unlock()
if retry >= len(b.ticks) {
return 0, false
}
ms := b.ticks[retry]
if b.jitter {
ms = jitter(ms)
}
return time.Duration(ms) * time.Millisecond, true
}
关键Backoff策略：
- ZeroBackoff: 不等待，立即重试。
- StopBackoff: 从不重试。
- ConstantBackoff: 固定等待时间。
- ExponentialBackoff: 指数增长的等待时间。
- SimpleBackoff: 提供一组固定的等待时间，可选择是否添加随机抖动。
- 锁
package lock
import (
"context"
"errors"
"fmt"
"time"
"github.com/gomodule/redigo/redis"
)
var (
// 防止孤儿lock没release
// 目前expire过期时间的敏感度是考虑为一致的敏感度
defaultExpireSecond uint32 = 30
)
var (
ErrLockSet = errors.New("lock set err")
ErrLockRelease = errors.New("lock release err")
ErrLockFail = errors.New("lock fail")
)
// RedisLockIFace 在common redis上封一层浅封装
// 将redis pool 与expire second作为redis lock已知数据
type RedisLockIFace interface {
MustSet(ctx context.Context, k string) (string, error)
MustSetRetry(ctx context.Context, k string) (string, error) // 必须设置成功并有重试机制
Release(ctx context.Context, k string, randVal string) error
}
// RedisLock nil的实现默认为true
type RedisLock struct {
redisPool *redis.Pool
expireSecond uint32
backoff Backoff
}
// An Option configures a RedisLock.
type Option interface {
apply(*RedisLock)
}
// optionFunc wraps a func so it satisfies the Option interface.
type optionFunc func(*RedisLock)
func (f optionFunc) apply(log *RedisLock) {
f(log)
}
// WithBackoff backoff set
func WithBackoff(b Backoff) Option {
return optionFunc(func(r *RedisLock) {
r.backoff = b
})
}
func NewRedisLock(redisPool *redis.Pool, opts ...Option) *RedisLock {
r := &RedisLock{
redisPool: redisPool,
expireSecond: defaultExpireSecond,
backoff: NewExponentialBackoff(30*time.Millisecond, 500*time.Millisecond), // default backoff
}
for _, opt := range opts {
opt.apply(r)
}
return r
}
func (r *RedisLock) Set(ctx context.Context, key string) (bool, string, error) {
if r == nil {
return true, "", nil
}
isLock, randVal, err := SetWithContext(ctx, r.redisPool, key, r.expireSecond)
if err != nil {
return isLock, randVal, ErrLockSet
}
return isLock, randVal, err
}
// MustSetRetry 必须设置成功并带有重试功能
func (r *RedisLock) MustSetRetry(ctx context.Context, key string) (string, error) {
op := func() (string, error) {
return r.MustSet(ctx, key)
}
notifyFunc := func(err error) {
if err == ErrLockFail {
fmt.Printf("RedisLock.MustSetRetry redis must set err: %v", err)
} else {
fmt.Printf("RedisLock.MustSetRetry redis must set err: %v", err)
}
}
return mustSetRetryNotify(op, r.backoff, notifyFunc)
}
func (r *RedisLock) MustSet(ctx context.Context, key string) (string, error) {
isLock, randVal, err := r.Set(ctx, key)
if err != nil {
return "", err
}
if !isLock {
return "", ErrLockFail
}
return randVal, nil
}
func (r *RedisLock) Release(ctx context.Context, key string, randVal string) error {
if r == nil {
fmt.Printf("that the implementation of redis lock is nil")
return nil
}
err := ReleaseWithContext(ctx, r.redisPool, key, randVal)
if err != nil {
fmt.Printf("s.RedisLock.ReleaseWithContext fail, err: %v", err)
return ErrLockRelease
}
return nil
}
func SetWithContext(ctx context.Context, redisPool *redis.Pool, key string, expireSecond uint32) (bool, string, error) {
if expireSecond == 0 {
return false, "", fmt.Errorf("expireSecond参数必须大于0")
}
conn, _ := redisPool.GetContext(ctx)
defer conn.Close()
randVal := time.Now().Format("2006-01-02 15:04:05.000")
reply, err := conn.Do("SET", key, randVal, "NX", "PX", expireSecond*1000)
if err != nil {
return false, "", err
}
if reply == nil {
return false, "", nil
}
return true, randVal, nil
}
func ReleaseWithContext(ctx context.Context, redisPool *redis.Pool, key string, randVal string) error {
conn, _ := redisPool.GetContext(ctx)
defer conn.Close()
luaScript := `
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end;
`
script := redis.NewScript(1, luaScript)
_, err := script.Do(conn, key, randVal)
return err
}
- 重试
package lock
import "time"
type mustSetOperation func() (string, error)
type ErrNotify func(error)
func mustSetRetryNotify1(operation mustSetOperation, b Backoff, notify ErrNotify) (string, error) {
var err error
var randVal string
var wait time.Duration
var retry bool
var n int
for {
if randVal, err = operation(); err == nil {
return randVal, nil
}
if b == nil {
return "", err
}
n++
wait, retry = b.Next(n)
if !retry {
return "", err
}
if notify != nil {
notify(err)
}
time.Sleep(wait)
}
}
- 使用
func main() {
backoff := lock.NewExponentialBackoff(
time.Duration(20)*time.Millisecond,
time.Duration(1000)*time.Millisecond,
)
redisPool := &redis.Pool{
MaxIdle: 3,
IdleTimeout: 240 * time.Second,
// Dial or DialContext must be set. When both are set, DialContext takes precedence over Dial.
Dial: func() (redis.Conn, error) {
return redis.Dial("tcp",
"redis host",
redis.DialPassword("redis password"),
)
},
}
redisLock := lock.NewRedisLock(redisPool, lock.WithBackoff(backoff))
ctx := context.Background()
s, err := redisLock.MustSetRetry(ctx, "lock_user")
if err != nil && err == lock.ErrLockFail {
fmt.Println(err)
return
}
time.Sleep(20 * time.Second)
defer func() {
_ = redisLock.Release(ctx, "lock_user", s)
}()
return
}
结论
通过深入理解Redis锁和Backoff重试策略的实现，我们可以构建出既能够保证资源访问的原子性，又能在面对网络波动或系统负载时保持稳定性的分布式锁机制。这不仅提高了系统的可用性，也增强了系统的容错能力。在实际开发中，合理选择和调整这些策略对于确保系统的高性能和高可靠性至关重要。通过精心设计的锁机制和重试策略，我们可以为分布式系统提供一个坚实的基础，以应对各种挑战和压力。
