Redis进阶：分布式锁实现

var (
	c chan struct{}
	r *redis.Client
)

func main() {
  // 连接 Redis
	r = redis.NewClient(&redis.Options{
		Addr: "127.0.0.1:6379",
	})
	r.Del("stock")   // 先清理之前的数据，为了方便测试

	chanCount := 5 // 协程数量
	c = make(chan struct{}, chanCount) // channel

	for i := chanCount; i > 0; i-- {
    // 五个协程
		go func() {
			for { // 不停地处理
				stock, err := r.Get("stock").Int()
				if err != nil { // 如果出错或者 Redis 中不存在 stock 就等着
					continue
				}
				if stock <= 0 { // 如果 stock 小于等于 0 就退出循环
					break
				}
				r.Decr("stock") // 扣减库存
			}
			c <- struct{}{} // channel 增加内容
		}()
	}
	
  // 等待 5 个 channel 处理完成
	for i := chanCount; i > 0; i-- {
		<-c
	}
}

127.0.0.1:6379> set stock 10000
OK

127.0.0.1:6379> get stock
"-1"

var (
	c chan struct{}
	r *redis.Client
	l sync.Mutex  // Go语言的锁
)

func main(){
  ……………………
  
  go func() {
			for {
				l.Lock()   // 加锁
				stock, err := r.Get("stock").Int()
				if err != nil {
					l.Unlock() // 有问题，释放锁
					continue
				}
				if stock <= 0 {
					l.Unlock() // 条件不满足了，释放锁，并结束循环
					break
				}
				r.Decr("stock")
				l.Unlock() // 完成操作，解锁
			}
			c <- struct{}{}
		}()
  ……………………
}

// 命令行1
go run main.go
// 命令行2
go run main.go
…………
// 命令行5
go run main.go

// Redis
127.0.0.1:6379> set stock 100000
OK

// 程序处理完成后再查看 Redis
127.0.0.1:6379> get stock
"-1"

  SETNX key value
  summary: Set the value of a key, only if the key does not exist
  since: 1.0.0
  group: string

……………………
l.Lock()
b := r.SetNX("lock", 1, 0) // 设置一个 SETNX 锁
if b.Err() != nil || !b.Val() { // 设置失败或者没有值的话，释放协程锁，并 continue 继续抢锁
  l.Unlock()
  continue
}

stock, err := r.Get("stock").Int()
if err != nil {
  r.Del("lock")  // 现在也要释放分布式锁了
  l.Unlock()
	continue
}
if stock <= 0 {
  r.Del("lock") // 现在也要释放分布式锁了
  l.Unlock()
	break
}
r.Decr("stock")
l.Unlock()
r.Del("lock") // 现在也要释放分布式锁了

……………………

………………
b := r.SetNX("lock", 1, 30*time.Second)
………………

………………
b := r.SetNX("lock", 1, 10*time.Second)
………………

var t *time.Timer
go func() {
  t = time.AfterFunc(10*time.Second/3, func() {
    fmt.Println("续命1次")
    r.Expire("lock", 10*time.Second)
  })
}()
rand.Seed(time.Now().UnixNano())
randomNum := rand.Intn(15) // 生成0~9的随机数
time.Sleep(time.Duration(randomNum) * time.Second) // 随机模拟耗时操作

stock, err := r.Get("stock").Int()
if err != nil {
  r.Del("lock")
  l.Unlock()
  continue
}
………………
// 注意下面所有删锁的地方也要停掉定时器
r.Del("lock")
t.Stop()
………………

// 进程一
  goprogromtourbook go run main.go
续命1次
8
续命1次
7
续命1次
4
1
续命1次
0
续命1次
续命1次
续命1次

// 进程二
➜  goprogromtourbook go run main.go
续命1次
9
续命1次
6
续命1次
5
3
2
续命1次
续命1次
续命1次
续命1次
续命1次

go func(index int) {
			lockId := strconv.Itoa(os.Getpid()) + ":" + strconv.Itoa(index)  // 简单做个标识
			…………………………
				if r.Get("lock").Val() == lockId {   // 要判断是自己的，才进行下面的解锁操作
					r.Del("lock")
					t.Stop()
					l.Unlock()
				}
			…………………………
}(i)

var (
	c     chan struct{}
	r     *redis.Client
	l     sync.Mutex
	cLock CLock
)

func main() {

	r = redis.NewClient(&redis.Options{
		Addr: "127.0.0.1:6379",
	})
	r.Del("stock", "lock", "verify")

	chanCount := 5
	c = make(chan struct{}, chanCount)

	for i := chanCount; i > 0; i-- {
		// 协程处理
		go decr(i)
	}

	for i := chanCount; i > 0; i-- {
		<-c
	}
}

func decr(i int) {
	// 初始化锁对象
	cLock = CLock{key: "lock", expire: 3 * time.Second}

	// 解锁操作函数
	allUnLock := func() {
		l.Unlock()
		cLock.UnLock()
	}

	// 异常处理
	defer func() {
		if err := recover(); err != nil {
			// 发生异常，解锁
			l.Unlock()
			allUnLock()
		}
	}()

	for {
		l.Lock() // 原生锁
		// 一致性
		lockId := strconv.Itoa(os.Getpid()) + ":" + strconv.Itoa(i)
		cLock.Lock(lockId) // 分布式锁

		// 续命
		go func() {
			cLock.lifeTimer = time.AfterFunc(cLock.expire-cLock.expire/3, cLock.Life)
		}()

		// 模拟耗时操作
		rand.Seed(time.Now().UnixNano())
		randomNum := rand.Intn(6)
		time.Sleep(time.Duration(randomNum) * time.Second)

		stock, err := r.Get("stock").Int()
		if err != nil {
			allUnLock()
			continue
		}

		if stock <= 0 {
			allUnLock()
			c <- struct{}{}
			break
		}

		r.Decr("stock")

		// 数据如果有重复操作，提示出来，并停止当前进程
		if res, _ := r.SAdd("verify", r.Get("stock").Val()).Result(); err != nil || res == 0 {
			allUnLock()
			fmt.Println("数据有重复: ", r.Get("stock").Val())
			os.Exit(1)
		}
		// 打印操作的是谁
		fmt.Println(lockId, "操作:", r.Get("stock").Val())

		allUnLock()
	}
}

type CLock struct {
	key       string        // 锁的 Key
	expire    time.Duration // 超时时间
	life      int           // 续命次数
	lifeTimer *time.Timer   // 续命定时器
	lockId    string        // 锁的值
}

func (cLock *CLock) Lock(id string) {
	for {
		b := r.SetNX(cLock.key, id, cLock.expire)
		if b.Err() != nil {
			fmt.Println(b.Err().Error())
			continue
		}
		if b.Val() {
			break
		}
	}
	cLock.life = 0
	cLock.lockId = id
	if cLock.lifeTimer != nil {
		cLock.lifeTimer.Stop()
	}
}

func (cLock *CLock) UnLock() {
	getId := r.Get(cLock.key).Val()
	if getId == cLock.lockId {
		r.Del(cLock.key)
		if cLock.lifeTimer != nil {
			cLock.lifeTimer.Stop()
		}
	}
}

func (cLock *CLock) Life() {
	if cLock.lockId != "" {
		if cLock.life < 3 {
			getId := r.Get(cLock.key).Val()
			if getId == cLock.lockId {
				r.Expire(cLock.key, cLock.expire)
				cLock.life++
				// cLock.lifeTimer.Stop()
				cLock.lifeTimer = time.AfterFunc(cLock.expire-cLock.expire/3, cLock.Life)
				fmt.Println(cLock.lockId, "续命", cLock.life, "次")
			}
		} else {
			panic("续命超次数")
		}
	}
}