基于redis实现秒杀
- 针对单进程实现
- 针对集群实现
- 1、redisTemplate.opsForValue().setIfAbsent方式
- 2、加入try{}finally{}、过期时间
- 3、redissonLock.lock()方式
- 4、读写锁（缓存与数据库读写不一致问题）
- redis优化方案
- zookeeper与redis实现分布锁区别
针对单进程实现
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21@RequestMapping("/miaosha1") @ResponseBody public String miaosha1() { synchronized (this){ Map<String,Object> mapListmaps = new HashMap<>(); //在redis中获取库存值 int kucun = Integer.valueOf(redisUtil.getStrKV("kucun").toString()); //如果库存大于0 if(kucun>0){ int newkucun = kucun - 1; //将库存值存入redis中（代表剩余库存） redisUtil.setStrKV("kucun",newkucun); System.out.println("剩余库存"+newkucun); }else{ System.out.println("库存不足"); } } return "list"; }
jdk提供synchronized方式单机情况下可用（单进程内置锁，只能控制当前进程），如果集群，就会存在并发问题
针对集群实现
1、redisTemplate.opsForValue().setIfAbsent方式
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28@RequestMapping("/miaosha2") @ResponseBody public String miaosha2() { String lockKey = "lockKey"; String clientId = UUID.randomUUID().toString(); //返回true代表保存成功 Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey,clientId); if(!result){ return "当前系统繁忙，请稍后再试"; } Map<String,Object> mapListmaps = new HashMap<>(); //在redis中获取库存值 int kucun = Integer.valueOf(redisUtil.getStrKV("kucun").toString()); //如果库存大于0 if(kucun>0){ int newkucun = kucun - 1; //将库存值存入redis中（代表剩余库存） redisUtil.setStrKV("kucun",newkucun); System.out.println("剩余库存"+newkucun); }else{ System.out.println("库存不足"); } //执行完之后，删除key（让下一个进来） redisTemplate.delete(lockKey); return "list"; }
redis分布式锁（入门版本）
用redisTemplate.opsForValue().setIfAbsent
如果出现相同的key值，redis默认会执行第一个，之后发现第二个和第一个key值相同，则第二个则不会执行
》》》缺点：如果redis数据减了，但是后续业务抛出异常（系统宕机），无法执行删除rediskey操作，后续就会死锁
2、加入try{}finally{}、过期时间
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39@RequestMapping("/miaosha3") @ResponseBody public String miaosha3() { String lockKey = "lockKey"; //uuid(为每一个锁添加一个唯一标识) String clientId = UUID.randomUUID().toString(); //返回true代表保存成功 //Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey,clientId); //添加超时间时间 Boolean result = redisTemplate.opsForValue().setIfPresent(lockKey,clientId,10, TimeUnit.SECONDS); if(!result){ return "当前系统繁忙，请稍后再试"; } try{ Map<String,Object> mapListmaps = new HashMap<>(); //在redis中获取库存值 int kucun = Integer.valueOf(redisUtil.getStrKV("kucun").toString()); //如果库存大于0 if(kucun>0){ int newkucun = kucun - 1; //将库存值存入redis中（代表剩余库存） redisUtil.setStrKV("kucun",newkucun); System.out.println("剩余库存"+newkucun); }else{ System.out.println("库存不足"); } }finally { //判断当前的redis是否是clientId下的redis if(clientId.equals(redisUtil.getStrKV("lockKey"))){ //执行完之后，删除key（让下一个进来） redisTemplate.delete(lockKey); } } return "list"; }
redis分布式锁（入门版本）
上一个缺点：如果redis数据减了，但是后续业务抛出异常，无法执行删除rediskey操作，后续就会死锁
》》》解决：加入try{}finally{}，不管执行过程如何都执行删除rediskey操作（）、（加入过期时间，避免系统宕机导致无法删除）
》》》缺点：因为时间问题，如果第一个耗时时间长，第一个过去10秒后超时将key删除，所以第二个进来了，但是第一个又走到finally将第二个key删除了，以此反复
》》》解决：为每一个锁加一个唯一标识（在finally之后判断当前的redis是否是clientId下的redis）
》》》问题：担心10秒过期时间超时，“锁续命”可以解决；设置一个定时任务查看锁是否执行完毕，如果未执行完毕，将10秒重置成10秒
》》》问题：分布式锁，释放锁得问题（原子性）
3、redissonLock.lock()方式
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30@RequestMapping("/miaosha4") @ResponseBody public String miaosha4() { String lockKey = "lockKey"; RLock redissonLock = RedisConfig.redisson().getLock(lockKey); try{ //加锁 redissonLock.lock(); //在redis中获取库存值 int kucun = Integer.valueOf(redisUtil.getStrKV("kucun").toString()); //如果库存大于0 if(kucun>0){ int newkucun = kucun - 1; //将库存值存入redis中（代表剩余库存） redisUtil.setStrKV("kucun",newkucun); System.out.println("剩余库存"+newkucun); }else{ System.out.println("库存不足"); } }finally { //释放锁 redissonLock.unlock(); } return "list"; }
redis分布式锁（中级版本）（公司可以够用了）
应用redisson加锁（redis升级版：如果加锁成功，每隔10秒检查”锁续命“；而且它本身自己会加标识，如果第一个锁未执行完毕，第二个不会进来）
》》》redissonLock.lock()源码：
lua脚本（里面内容就是上一个步骤：加key，加标识，加超时时间，还有一个锁续命的逻辑）
》》》问题：主从节点同步问题（第一个redis加key值时，会先到主节点，再到从节点。但是现在主节点挂了，第一个无法完成。这时候第二个又进来了）
4、读写锁（缓存与数据库读写不一致问题）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37@RequestMapping("/miaosha5") @ResponseBody public String miaosha5() { String lockKey = "lockKey"; //读写锁 RReadWriteLock rReadWriteLock = RedisConfig.redisson().getReadWriteLock(lockKey); //读 RLock writeLock = rReadWriteLock.writeLock(); //写 RLock rLock = rReadWriteLock.readLock(); try{ //读加锁 writeLock.lock(); //写加锁 rLock.lock(); //在redis中获取库存值 int kucun = Integer.valueOf(redisUtil.getStrKV("kucun").toString()); //如果库存大于0 if(kucun>0){ int newkucun = kucun - 1; //将库存值存入redis中（代表剩余库存） redisUtil.setStrKV("kucun",newkucun); System.out.println("剩余库存"+newkucun); }else{ System.out.println("库存不足"); } }finally { //释放锁 writeLock.unlock(); rLock.unlock(); } return "list"; }
问题：缓存与数据库读写不一致问题
1、更新完数据库删缓存；另一个线程卡顿
2、延迟双删（sleep多少毫秒）；另一个线程卡顿时间不好估计，
3、内存队列（数据操作直接到一个队列中，原子性）
4、加分布式锁（每一段操作）
4.5、读写锁（分布式读锁和写锁，读写有关联，读跟读无关联）----源码：读写锁key是相同的，区别是加一个模式read、write
》》》场景：购买商品时，因为期间会有很多步骤“0加购物车、付款等”到真正付款时缓存库存值清零了，缓存不一致问题，
》》》解决方案：设置一个超时时间，expire（30s）；
读多写多用缓存不太好，命中率降低，也可以用canal，canal中间件可以监控数据库执行的先后顺序，然后再更新缓存
redis优化方案
1、较少锁粒度，在锁内尽可能减少代码量，
2、分段锁：将库存分为几个库存，让同时进行这个扣减，增加效率
zookeeper与redis实现分布锁区别
保存数据上，redis主节点保存之后直接会回调程序保存成功，zk是主节点保存之后会让自己余下的从节点比如3个，保存2个1（一半以上）才会回复主程序
保存成功。安全性比redis高，但是速度慢
》》》redis有类似zk的方式解决方式：Redlock（小问题比较多）；执行方式就是用三个redissonLock.lock();加锁，然后用Redlock控制
最后
以上就是碧蓝跳跳糖最近收集整理的关于基于redis实现秒杀针对单进程实现针对集群实现redis优化方案zookeeper与redis实现分布锁区别的全部内容，更多相关基于redis实现秒杀针对单进程实现针对集群实现redis优化方案zookeeper与redis实现分布锁区别内容请搜索靠谱客的其他文章。
发表评论 取消回复