网站推广计划书具体包含哪些基本内容?,wordpress 不做SEO,做ppt什么网站图片好,电子商务网址大全Redis进阶#xff1a;分布式锁实现 锁这个概念#xff0c;不知道大家掌握的怎么样。我是先通过 Java #xff0c;知道在编程语言中是如何使用锁的。一般 Java 的例子会是操作一个相同的文件#xff0c;但其实我们知道#xff0c;不管是文件#xff0c;还是数据库中的一条… Redis进阶分布式锁实现 锁这个概念不知道大家掌握的怎么样。我是先通过 Java 知道在编程语言中是如何使用锁的。一般 Java 的例子会是操作一个相同的文件但其实我们知道不管是文件还是数据库中的一条数据只要是多个进程或者线程共享的需要共同操作的数据都会因为异步读写而带来一些问题这些问题最终的解决方案往往就是通过锁。 MySQL 中有表锁、行锁、读写锁、意向锁、间隙锁等等而在 Redis 中其实并没有完全的锁的概念因为它是单线程执行命令的本身就不会出现多线程同时操作同一数据的问题。因此使用 Redis 非常适合来做分布式锁。那么分布式锁又是啥意思呢 单进程单线程或者只在一台主机上其实我们有很多种锁的解决方案。但是如果跨进程、跨主机了不借助外力去实现锁就非常麻烦了。在同一个程序中我们直接使用程序代码提供的锁就可以了比如 Go 中的 sync.Mutex 而如果不在同一个进程中显然这个锁是没法让另一个进程也锁上的。这时我们可以用类似于 pid 文件这样的概念就是上锁的时候创建一个文件这个文件即使是空的也行。其它进程如果发现有这样一个文件那么就表示有别的程序已经上锁了。这是非常简单地解决同一台机器下多个进程之间锁的方案。而如果跨多台机器的多台实例那么大部分情况下我们都会使用一些第三方的存储。MySQL 可以做分布式锁吗可以的加锁的时候插入一条数据解锁的时候删除。Redis 可以吗可以而且更方便那么咱们就来好好聊聊 Redis 的分布式锁。另外Zookeeper 也是现在非常流行的分布式锁方案的一种外部存储实现。 普通的线程协程锁 这里我们先来看一下在一个普通的程序中如何对多个线协程加锁今天的内容为了方便起见我使用的是 Go 语言如果是 PHP 的话Swoole 中也有类似的锁大家可以自己尝试。 先看问题假设做一个秒杀场景最典型的问题就是超卖问题意思就是多协程或者分布式多实例的情况下同时操作库存字段有可能让库存变成负的也就是出现了超卖的情形。 var (c chan struct{}r *redis.Client
)func main() {// 连接 Redisr redis.NewClient(redis.Options{Addr: 127.0.0.1:6379,})r.Del(stock) // 先清理之前的数据为了方便测试chanCount : 5 // 协程数量c make(chan struct{}, chanCount) // channelfor 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}
} 代码不多解释了如果没有学过 Go 但学过 Swoole 的话应该大体也能看懂如果都不会额赶紧去学一下吧。接下来我们往 Redis 中添加数据比如增加 10000 个库存。 127.0.0.1:6379 set stock 10000
OK 程序那边很快就会执行完成这时我们再来看看库存还有多少。 127.0.0.1:6379 get stock
-1 不对呀明明我们判断了 如果 stock 小于等于 0 就退出循环 这个条件为啥会出现 -1 这就是多线协程会出现的典型问题了。有的时候还不一定只会是 -1 还会出现 -2 、-3 等等状况。如果在真实的业务场景中只是超卖了这么几个倒还好说但秒杀场景下往往是超大的流量和超大的并发万一超卖了几百上千件的话.....看甲方爸爸怎么收拾你。 我就简单地画了两个协程可能出现的问题如果你学过 Java 了解多线程的话把 Goroutine 看成是 Thread 也是可以的。对于同一个程序来说这个问题好处理加个锁就好了。 var (c chan struct{}r *redis.Clientl 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{}{}}()……………………
} 标准处理流程不解释了现在在单应用实例的情况下其实就已经没有太大问题了你可以自己试试把 stock 的数量再加大或者再多加几个协程也都是没问题的。如果想提高性能可以再换成读写锁这里就不演示了。接下来我们通过运行多个应用实例模拟分布式状态下又会出现什么问题。 // 命令行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 我去怎么又超卖了 分布式锁 实际上根源和上一张图的原因是类似的进程内部的锁只能保证当前这个进程内的线协程上锁其它的进程或者真正的分布式放在其它的主机上这玩意肯定不起效果了。这时往往就需要借助第三方来实现分布式的锁了。 在 Redis 中我们已经知道它的命令执行就是单线程的所以这一块我们不用太操心因此我们只需要一个高大上的而且是非常简单的在第一课就提到过的命令就可以实现分布式锁它就是 SETNX 命令。 SETNX key valuesummary: Set the value of a key, only if the key does not existsince: 1.0.0group: string 这个命令的意思是如果键存在就无法设置它。既然这样的话那么我们上锁的时候去 SETNX 一下如果成功了就表明当前的进程拿到了锁如果失败了就等待继续抢锁。操作执行完成之后直接 DEL 设置的那个值就好了这不就实现了一个分布式锁嘛。话不多说继续改造代码。 ……………………
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) // 现在也要释放分布式锁了…………………… 就加了这么几行代码其实分布式锁的大体轮廓就已经出来了。测试一下吧现在处理 10 万条库存的话即使 5 个进程每个进程有 5 个协程速度也会变得非常慢了。这是为啥呢其实任何锁的操作都是在做一件事把并行变成串行让同时操作变成顺序操作。目的就是用速度换正确性或者说是牺牲效率换一致性。 也就是说一步一步的加锁过来我们的程序在处理高并发的业务的能力也在一步步的下降如果再说锁的性能优化那又是各种长篇大论了也不是我们今天讨论的范围。但是还是给大家提供一些思路比如读写锁实际的业务开发中往往不像我们这样测试的这么极端还会有别的操作比如可能先读到数据进行一些其它处理最后才写这时就可以上读写锁来提高性能即在写锁之前读锁获取到的内容还是可以进行其它操作的。 好了话题拉回来今天我们还是着重解决最核心的分布式锁的问题性能问题将来有机会再探讨。现在这个分布锁的应用就 OK 了完全没问题了不不不还有几个问题值得我们考虑一下。 如果一个进程卡死了没法释放锁了咋办如果一个进程可能某些原因确实要执行一段时间如何释放锁比较合适有没有可能某一个进程释放了另一个进程的锁 这三个问题也是面试时非常常见的问题解决的思路其实也并不复杂。 进程卡死了没法释放锁SETNX 的时候加个过期时间呗进程确实可能执行时间长增加续命功能就是快到期前如果锁还没释放就给它加长点时间别的进程释放了我的锁给设置的 Value 指定一个明确的内容可以是客户端ID或者进程机器ID之类的 设置一个过期时间 这个不用多解释了吧这里我直接设置了 30 秒就算进程挂掉了30 秒后这个锁也会自动释放。 ………………
b : r.SetNX(lock, 1, 30*time.Second)
……………… 续命 这个问题不是太好测我们可以先把过期时间调短点比如上面的 30 秒换成 10 秒。然后进行续命操作。 ………………
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()
……………… 这里我们为什么要新开一个协程呢主要就是不影响主协程的任务处理让新的协程在定时器的时间到达之后进行续命其实也就是增加操作的时间。然后我们就开两个进程进行测试由于暂停时间是随机的所以我们也少给点数据就上 10 个库存好了。 // 进程一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次 其实不用我多说你也会发现问题那就是万一某个操作一直卡住了锁还是无法释放啊所以正式情况下我们还要考虑续命次数一般来说续个2、3次还不行的话可以认为当前这个操作是有问题的应该采取别的方案比如报错抛异常之类的。 只删自己的锁误删锁问题 考虑一个场景比如 A 、B 两个进程都是耗时任务A 上锁了过期了不考虑多次续命抛出异常强制结束的情况锁被自动释放了B 又过来上锁了正在处理的过程中 A 那边突然炸尸了结束任务把锁又释放了这时不就是 A 把 B 的锁释放了嘛然后其它进程 C、D、XXX 都来抢锁马上就进入混乱状态了。 好解决吗说难也难说不难也不难。说不难是因为我们可以使用 value 值啊设置一个标识当前进程状态的内容就可以了删除的时候只有匹配上了才能删自己的。 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) 说难吧就是这个标识的唯一性以及这个思想的转变我们的测试其实不是很好测出释放了别人锁的情况但如果是一个合格的架构师应该是要去考虑这一块的内容的而且这个也是非常容易想到的可能出 BUG 的一个点。 完整代码 其实啊上面这堆东西大部分是面试才用得到为啥这么说呢真正的高并发大流量的系统百分之90以上全是 Java 了而 Java 中的 Redisson 包中有非常完整的这套分布式锁方案完全不用自己写。不知道 Go 和 PHP 有没有现成的类似于 Redisson 的包如果有了解的小伙伴评论区留下地址我们一起去学习哦。 下面就把稍微优化过后的一整套完整的代码放上来大家可以自己运行测试一下哦。具体解释记得之后看视频。 var (c chan struct{}r *redis.Clientl sync.MutexcLock CLock
)func main() {r redis.NewClient(redis.Options{Addr: 127.0.0.1:6379,})r.Del(stock, lock, verify)chanCount : 5c 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 // 锁的 Keyexpire 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 0cLock.lockId idif 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(续命超次数)}}
} 上述代码并不是完全的真正可用的分布式锁代码只是我们根据分布式锁的概念去写的并没有进行详细的测试所以可能还是会有很多的 BUG 和问题不推荐直接复制走了就放到线上去使用哦。 红锁 RedLock 红锁这个概念啊其实和 Zookeeper 还有点关系另外还要和存储架构中的 CAP 原理扯上关系。CAP 表示的就是一致性、可用性和分区性当分区性存在的时候时候CA 无法两全。啥意思一旦有分区可用和一致就没办法一起达到毕竟只要有分区就一定会有延迟不管延迟的大小它一定是客观存在的。想要高可用那就尽量减少分布数量或者仅从一台主机获取结果就认为结果是确定的。而想要一致性也就是所有主机的数据都确定或者超过半数以上的主机确认这条数据才真正的确定是当前的值这样可用性就差了为啥要一条一条的每台机器都确认嘛。 正常来说Redis 的分布式锁是 AP 式的而 Zookeeper 的分布式锁是 CP 式的。一般 Zookeeper 都是三台机器起步同一把锁必须有两台主机确认锁上了它才会返回真的锁上了。而在 Redis 中其实没有这样的机制即使是 Cluster 也是把 Key 做 Hash 放到不同的机器上去了。这也是 Redis 的特点所决定的它本身就是一个高可用的快速缓存系统嘛。 不过各位大佬们不甘心于是就出现了 RedLock 这个概念。其实就是多台 Redis 通过程序代码控制比如配置三台 Redis 主机上锁的时候必须有两台上锁成功了才表示成功。 同样Java 的 Redisson 中也有现成的解决方案对于我们来说知道有这么个东西就好了。自己想要去实现的话最好还是看看分析一下 Redisson 的源码然后去依照它的写就像上面的我们实现的分布式锁也是一样实验性质的和 Java 中已经成熟的工程化的代码还是有很大区别的。毕竟它们会考虑更多的问题而我们可能有很多会想不到。 在 PHP 中其实通过 packagist.org 还是能搜到不少的 RedLock 相关的组件的只不过我并没有试过有没有好用的大家有用到的朋友可以给点意见哈。 总结 今天的内容有点长但其实主要是代码粘的多点。再次强调我们最后的分布式锁的代码只是测试性质的看了很多讲 Redis 的教程中各位老师基本都会用 Java 来实现一个简单的分布式锁因此我这个其实也大部分是根据分布式锁的概念用 Go 代替 Java 写了一遍有会 Swoole 的小伙伴也可以尝试改造成 Swoole 版本的哦。但是就像他们的课程最后都会引到 Redisson 上一样确实自己写的会有很多地方欠缺考虑专业开源稳定的一些软件包才是真正应用在业务开发中的首选。如果真的想用其实不太需要考虑续命之类的问题这一块要完美实现真的很绕的只要超时没执行完就中断好了。这样一个简单的分布式锁其实也足够应对不小的流量和大部分的需求了。要知道能力越大责任越大有时候如果我们无法考虑的完美那么干脆换个角度看能不能不要这个特性通过其它更简单的方式来弥补。 可惜的是在 PHP 和 Go 领域貌似还没发现有类似 Redisson 这样的非常出名的 Redis 分布式扩展去搜索一下也能找到一些至于好不好用我也不清楚。还是那句话平常用不到啊大家加油冲向大厂啊努力找大神带带啊要不这些玩意学起来真 TNND 的费劲。 测试代码 https://github.com/zhangyue0503/dev-blog/blob/master/redis/2022/source/25.go
