首页 > 数据库 >深入理解Redis锁与Backoff重试机制在Go中的实现

深入理解Redis锁与Backoff重试机制在Go中的实现

时间：2024-09-16 09:52:27浏览次数：10  
标签：return string err redis Redis Backoff 重试 func time

目录

在构建分布式系统时，确保数据的一致性和操作的原子性是至关重要的。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方法成功获取锁或达到最大重试次数。通过这种方式，我们能够确保即使在高竞争环境下，也能以一种可控和安全的方式获取锁。

具体实现

  1. 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: 提供一组固定的等待时间，可选择是否添加随机抖动。
  1. 锁
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
}
  1. 重试
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)
	}

}
  1. 使用

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重试策略的实现，我们可以构建出既能够保证资源访问的原子性，又能在面对网络波动或系统负载时保持稳定性的分布式锁机制。这不仅提高了系统的可用性，也增强了系统的容错能力。在实际开发中，合理选择和调整这些策略对于确保系统的高性能和高可靠性至关重要。通过精心设计的锁机制和重试策略，我们可以为分布式系统提供一个坚实的基础，以应对各种挑战和压力。

标签：return,string,err,redis,Redis,Backoff,重试,func,time
From： https://www.cnblogs.com/crazytata/p/18416026

相关文章

  • redis主从环境搭建
    准备系统环境：Linuxubuntu4.15.0-62-generic下载redis-6.2.4.tar.gz(官网下载即可)解压编译redistar-xvfredis-6.2.4.tar.gzcdredis-6.2.4make环境准备三个redis服务，一主二从(安装在本地所以用127.0.0.1，此版本从服务器用replica)：127.0.0.1:6379master127.......
  • 编程日记 后端使用redis
    编程日记后端使用redis问题：在写伙伴匹配系统的时候，要使用redis存储数据，需要在xml文件中设置spring.session.tore-type=redis但是3.x版本的spring不再有这个参数。那么怎么设置是否使用redis存储呢？只要导入了org.springframework.session包那就启用，不导入就不启用https://www.......
  • 一、redis简介
    redis是当今使用最广泛的分布式缓存。官网是：https://redis.io/。还有其他的缓存中间件，比如memcache，GuavaCache等。那么redis的优势在哪呢？redis是内存数据库，数据主要在内存中。redis可通过持久化策略将数据保存到文件中。redis有丰富的数据类型：string，最常用的类型list，可......
  • Redis 管道技术
    概述Redis是基于请求/响应协议的TCP服务。在客户端向服务器发送一个查询请求后，需要监听Socket的返回结果，该监听过程一直阻塞，直到服务器有结果返回。如果一次请求延迟20ms，则多次请求的网络延迟会不断累加。也就是说，Redis的性能瓶颈主要体现在网络延迟上Redis的管道技术指......
  • redis基本数据结构-set
    文章目录1.set的基本介绍1.1.set底层结构之hash表的简单介绍1.2.常用命令2.常见的业务场景2.1.标签系统2.2.社交网络好友关系1.set的基本介绍参考链接：https://mp.weixin.qq.com/s/srkd73bS2n3mjIADLVg72Aredis的set数据结构是一个无序的集合，可以存储不......
  • k8s部署单机redis pod
     kind:ConfigMapapiVersion:v1metadata:name:redisnamespace:defaultlabels:app:redisdata:redis.conf:|-port6379bind0.0.0.0dir/dataappendonlyyesprotected-modenorequirepassredispidfile/data/red......
  • 云服务器安装redis
    第一步：上传redis压缩安装包到服务器       wgethttps://download.redis.io/releases/redis-5.0.4.tar.gz第二步：将压缩安装包解压        tar-xvfredis-xxx.tar.gz第三步：进入redis的目录，编译redis，执行命令：make        cdr......
  • Redis实现发布/订阅功能（实战篇）
    前言博主在学习Redis实现发布订阅功能的时候，踩了太多的坑。不是讲解不详细，看的一知半解；就是代码有问题，实际压根跑不起来！于是博主萌生了自己写一个最新版且全程无错的博客供各位参考。希望各位不要把我才过的坑再踩一遍。（实战篇的所有代码均由本人测试，全程无Bug。）废话不多......
  • Redis运维之监控指标，性能监控，监控方式，响应慢分析
    目录1Redis监控1.1Redis监控指标1.1.1性能指标:Performance1.1.2内存指标:Memory1.1.3基本活动指标：Basicactivity1.1.4持久性指标:Persistence1.1.5错误指标：Error1.2监控方式1.2.1info1.2.2性能监控1.2.3内存监控1.2.4基本活动指标1.2.5持久性指标1.2.6错误指标......
  • Redis集群
    Redis集群一、定义由于数据量过大，单个master复制集难以承担，因此需要对多个复制集进行集群，形成水平扩展，每个复制集只负责存储整个数据集的一部分，这就是redis的集群。Redis集群是一个提供在多个redis节点间共享数据的程序集，可以支持多个masterRedis集群支持多个master，每个mast......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99