参考

Redis基础 - 基本类型及常用命令
Redis基础 - Java客户端
Redis 基础 - 短信验证码登录
Redis 基础 - 用Redis查询商户信息
Redis 基础 - 优惠券秒杀《非集群》

synchronized在集群上使用时的问题

synchronized只能够保证一个JVM内部的多个线程之间的互斥，而无法在集群之间互斥，要想解决这个问题必须要使用分布式锁。分布式锁是满足分布式系统或集群模式下“多进程可见”并且能“互斥”的锁。

为何需要分布式锁

比如有两个JVM，JVM1和JVM2，synchronized就是利用JVM内部的锁监视器来控制线程，在JVM的内部因为只有一个锁监视器，所以只会有一个线程获取锁，因此可以实现线程之间的互斥。但当有多个JVM的时候，就会有多个锁监视器，那么就会有多个线程获取到锁，这样的话无法实现多JVM之间的互斥。要想解决这个问题，肯定不能使用JVM内部的锁监视器了，必须让多个JVM去使用同一个锁监视器。所以他一定是在JVM外部的，多JVM进程都可以看到的锁监视器，这时候无论是JVM内部的还是多JVM的线程，都应该去找外部的锁监视器获取锁，这样也就会只有一个线程获取锁，就能实现多进程之间的互斥了。

业务场景

比如JVM1里有线程1在执行业务，她就会去获取互斥锁，她获取锁就会去找外部的锁监视器，一旦获取成功，就在锁监视器里记录当前获取锁的是线程1。此时如果其他线程也来获取锁，比如JVM2内部的线程3，她也会去外部的锁监视器试图获取锁，但因为锁监视器已经有线程1使用着，所以线程3获取一定会失败，失败之后她就会去等待锁释放。一方面，假如JVM1的线程1执行着：先查询订单，若没有就插入新订单，由于她是第一个来的，所以没有订单，所以插入新订单，执行完后就会释放锁。等线程1释放完之后，线程3拿到锁了，她也去执行一样的业务：获取锁成功，查询订单，但查询时由于线程1已经插入了，所以线程3就能查询到订单，由于已经存在，所以直接返回报错。

分布式锁的实现

分布式锁的核心是实现多进程之间互斥，而满足这一点的方式有很多，常见的有三种：

	MySQL	Redis	zookeeper
互斥	利用民事权利本身的互斥锁机制互斥	利用setnx这样的互斥命令	利用节点的唯一性和有序性实现互斥
高可用	好	好	好
高性能	一般	好	一般
安全性	断开连接，自动释放锁	利用锁超时时间，到期释放	临时节点，断开连接自动释放

基于Redis的分布式锁

实现分布式锁时需要实现的两个基本方法：

（1）获取锁

互斥：确保只能有一个线程获取锁，可以利用setnx。为了不让setnx后还没来得及设置expire时恰巧宕机，要保证setnx和expire同时成功或同时失败，所以可以用set命令，set命令有很多参数，比如set key value EX 10 NX，即设置了值还设置了EX（超时时间）参数，为10秒，还配了NX参数（互斥），没值时才设置。利用这个命令可以让NX和EX变成原子操作。
非阻塞方式去获取锁：尝试一次，成功返回true，失败返回false

（2）释放锁

手动释放：del 锁的key
超时释放：获取锁时添加一个超时时间，即expire。避免服务宕机而出现的死锁。

基于Redis实现分布式锁（初级）

ILock.java

public interface ILock{// 尝试获取锁（因为这里用的是非阻塞获取）/*参数：锁持有的超时时间，过期后自动释放返回值：true 获取锁成功； false 获取锁失败*/boolean tryLock(long timeoutSec);// 释放锁void unlock();
}

SimpleRedisLock.java

public class SimpleRedisLock implements ILock {public String name;// 业务名，不同的业务的锁的key要不同（实际上就是key名）。private StringRedisTemplate stringRedisTemplate;// 通过构造函数传值public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}private static final String KEY_PREFIX = "lock:";//key的前缀@Overridepublic boolean tryLock(long timeoutSec) {// 获取当前线程的id（线程的标识）long threadId = Thread.currentThread().getId();// 获取锁，setIfAbsent 如果不存在则执行。value最好加个标识，哪个线程拿到的锁。Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId + "", timeoutSec, TimeUnit.SECONDS);// 返回值是boolean基本类型，而这里是Boolean，所以直接返回的话会自动拆箱，如果值是null，拆箱时可能会报空指针错误，所以返回时可以这么返回，防止自动拆箱return Boolean.TURE.equals(success);// Boolean.TURE是常亮，所以与success比较后，一样就返回true，不一样（或是null时）就返回false。}@Overridepublic void unlock() {stringRedisTemplate.delete(KEY_PREFIX + name);}
}

代码示例1：用一人一单的例子来测试

VoucherOrderServiceImpl.java

@Resource
prviate ISeckillVoucherService iSeckillVoucherService;@Resource
prviate RedisIdWorker redisIdWorker;@Resource
prviate StringRedisTemplate stringRedisTemplate;@Override
// @Transactional 由于这个方法里只有查的，所以不需要这个。
public Result seckillVoucher(Long voucherId) {// 1，根据voucherId查询优惠券SeckillVoucher voucher = seckillVoucherService.getById(voucherId);// 秒杀券的id也和优惠券的id值是共有的，所以能这么查// 2，判断秒杀是否开始// 判断开始时间是不是大于现在时间if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {// 尚未开始return Result.fail("秒杀尚未开始");}// 3，判断秒杀是否已经结束// 结束时间是不是当前时间的之前if (voucher.getEndTime().isBefore(LocalDateTime.now())) {// 已经结束return Result.fail("秒杀已经结束");}// 4，判断库存是否充足if (voucher.getStock() < 1) {// 库存不足return Result.fail("库存不足");}Long userId = UserHolder.getUser().getId(); // 用户id/****************** 修改的部分 *********************/// 创建锁的对象/*关于锁的名称的建议：一般取跟你的业务有关的标识，这个是下单的业务，所以可以用“order”，还需要注意一点锁的范围，如果只写成order的话，意味着凡是来下单的业务都会被锁定，但这里我们要锁定的范围是用户，即同一个用户我们才要加限制，不同的用户无所谓，所以这里的锁的范围应该是用户，所以拼接userId。*/SimpleRedisLock lock = new SimpleRedisLock("order"+userId , stringRedisTemplate);// 尝试获取锁boolean isLock = lock.tryLock(10);// 时间跟业务执行时间有关，一般5秒或10秒就行，因为这时间内业务都能执行完/****************************************************/// 判断是否获取锁成功if (!isLock) {// 若不成功，可以返回或重试（只是这里是要防止一个用户重复下单，所以这个业务下是不能“重试获取锁”的）return Result.fail("一个人只允许下一单");}try {// 拿到当前对象的代理对象（获取跟事务有关的代理对象）IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();/* 当然，这个函数createVoucherOrder只存在实现类里，所以在接口里也要创建。另外，这么做的话，底层还要依赖aspectj，所以要在pom.xml里添加相关依赖。还要在启动类暴露这个代理对象*/return proxy.createVoucherOrder(voucherId);} finally {lock.unlock();}
}@Transactional // 更新的都跑到这里了，所以这里加事务
public Result createVoucherOrder (Long voucherId) {// 5，一人一单// 5.1 查询订单int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count();// 5.2 判断是否存在if (count > 0) {// 用户已经购买过了return Result.fail("该用户已经购买过一次");}// 6，更新库存boolean success = seckillVoucherService.update().setSql("stock = stock - 1") // set stock = stock - 1.eq("voucher_id", voucherId).gt("stock", voucher.getStock()) // where id = ? and stock = ?.update();if (!success) {// 库存不足return Result.fail("库存不足");}// 7，创建订单VoucherOrder voucherOrder = new VoucherOrder();// 对应着优惠券订单表tb_voucher_order// 7.1 订单idlong order_id = redisIdWorker.nextId("order");voucherOrder.setId(order_id);// 7.2 用户idvoucherOrder.setUserId(userId);// 7.3 代金券idvoucherOrder.setVoucherId(voucherId);// 7.4 其他的是取默认值，所以不用设置// 7.5 订单信息写入数据库save(voucherOrder);// 8，返回订单idreturn Result.ok(order_id);
}

在大多数情况下，上面的代码不会出问题，但一些极端的情况下，还是会出现问题。

Redis分布式锁误删问题

比如，有业务1，获取锁之后，执行业务，但出现了业务阻塞情况，就导致业务执行时间超过了锁的超时时间，这时候就会触发锁的超时导致的释放，问题是这时候业务还没有完成呢锁就提前释放了。一旦锁提前释放，这时候比如线程2再来获取锁的时候，就能趁虚而入，获取锁成功，之后线程2也会执行自己的业务，而就在线程2刚刚获取锁了以后，假设线程1醒了，业务也完成了，所以然后线程1要释放锁了，即直接del key了，于是，由于这时候是线程2在业务执行中，所以是线程2的锁被释放了，但线程2不知道这些，线程2还在去执行自己的业务，就在这时，线程3来了，她也趁虚而入，也获取了锁，由于锁刚才被线程1给删了，所以线程3也能获取成功，也执行自己的业务，此时此刻，同时由两个线程都拿到了锁即线程2和线程3，他们都在执行着业务，所以又一次出现了并行执行的情况，那么线程安全的问题就有可能再次发生。（网友1：这是同一用户在不同客户端访问服务器的情况）出现这种情况的原因是，第一由于出现业务阻塞，导致锁提前释放，第二是当线程1醒过来以后，这时候的锁已经不是线程1的锁了，而是线程2的锁，但线程1二话不说上来就把别人的锁给干掉了。所以归根结底，发生这错误的最重要的原因是：线程1释放锁时把别人的锁给干掉了。

所以在释放锁的时候，判断一下锁的标志是否跟当前线程（比如这里是线程id）一致。

网友之评

线程1和线程2不也是并行？
那超时的怎么办？

看来很多人都共同关心这个问题，既然讲者没解释这个问题，干脆把超时设的较长点得了，其余的就听天由命吧。

解决误删问题的办法

1，在获取锁时存入线程标识（可以用UUID表示）
上面例子中用的是线程id，线程id是递增的数字，在JVM内，每创建一个线程，她就会递增。所以，如果是在集群模式下，有多个JVM，每个JVM内部都有维护递增的数字，所以两个JVM很有可能出现线程id冲突的情况。

2，在释放锁时先获取锁中的线程标示，判断是否与当前线程标示一样

如果一致则释放锁
如果不一致则不释放锁

代码示例2：修改锁的实现类

SimpleRedisLock.java

public class SimpleRedisLock implements ILock {public String name;// 业务名，不同的业务的锁的key要不同（实际上就是key名）。private StringRedisTemplate stringRedisTemplate;// 通过构造函数传值public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}private static final String KEY_PREFIX = "lock:";//key的前缀private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";// 使用hutool工具类的，传true就能去掉横线，比较方便@Overridepublic boolean tryLock(long timeoutSec) {// 获取当前线程的id（线程的标识），前面拼接UUID，保证集群下的唯一String threadId = ID_PREFIX + Thread.currentThread().getId();// 获取锁，setIfAbsent 如果不存在则执行。value最好加个标识，哪个线程拿到的锁。Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);// 返回值是boolean基本类型，而这里是Boolean，所以直接返回的话会自动拆箱，如果值是null，拆箱时可能会报空指针错误，所以返回时可以这么返回，防止自动拆箱return Boolean.TURE.equals(success);// Boolean.TURE是常亮，所以与success比较后，一样就返回true，不一样（或是null时）就返回false。}@Overridepublic void unlock() {// 获取当前线程标示String threadId = ID_PREFIX + Thread.currentThread().getId();// 获取锁中的标示String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);// 就是Redis中的key的值// 判断标示是否一致if (threadId.equals(id)) {// 一致时才删stringRedisTemplate.delete(KEY_PREFIX + name);}}
}

经过这么修改，在大多数情况下都能正常，但在极端情况下依然会出现问题。

原子性问题

比如，线程1去获取锁，由于是她自己，所以一定能成功，于是她开始执行自己的业务，假设这个业务并没有阻塞，她成功执行完了，紧接着她要去释放锁了，于是要判断锁标示是否和自己的一样，由于此时锁是她自己的，所以这个判断一定是一样的，紧接着她就要释放锁了，这里的判断锁标示和释放锁是两个动作，判断是成功了，紧接着要进行释放，而就在要释放时，产生了阻塞（为何可能会出现阻塞呢，在JVM里有个东西是垃圾回收，当JVM去做FULL GC的时候，她就会阻塞我们的所有的代码，所以这个时候就会产生阻塞，即不是因为业务阻塞，而是因为JVM本身阻塞），一旦发生了阻塞，而现在是轮到线程1去释放锁了，但是由于被阻塞，所以锁没能被释放，而这个阻塞的时间如果足够长，很有可能会触发锁的超时释放，一旦锁被超时释放，其他的线程又可以趁虚而入了，比如此时线程2过来获取锁，因为锁刚才被超时释放，所以线程2就能成功获取锁，于是她开始执行自己的业务，而就在他获取锁成功的那一刻，假如GC结束了，那么阻塞结束了，线程1恢复运行，而此时她要执行释放锁的动作了，因为锁是否一致的判断已经执行过了，所以她认为锁还是自己的（网友1感叹：太极端了），但其实现在的锁是线程2的，所以线程1就会直接执行释放锁（网友2感叹：考虑的情况好多呀），于是就把线程2的锁给干掉了。又一次发生了误删。那么此时又来线程3趁虚而入，获取锁成功，执行自己的业务，这种并发的问题又一次发生了。

所以，要想避免这个问题的发生，必须确保判断锁标示是否一致的动作和释放锁的动作，这两个要整成一个原子性的操作，也就是说，一起执行，不能出现间隔。

lua脚本解决多条命令原子性问题

Redis提供了lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。

lua是一种编程语言，她的基本语法可以参考：https://www.runoob.com/lua/lua-tutorial.html

这里重点介绍Redis提供的调用函数，语法如下：

#执行redis命令
redis.call('命令名称', 'key', '其他参数', ...)

例如，我们要执行set name jeck，则脚本是这样：

redis.call('set', 'name', 'jack')

例如，我们要先执行set name rose，再执行get name，则脚本如下：

redis.call('set', 'name', 'jack')
local name = redis.call('get', 'name')
return name

写好脚本以后，需要用Redis命令来调用脚本，调用脚本的常见命令如下：

EVAL script numkeys key [key ...] arg [arg ...]

例如，我们要在终端执行 redis.call(‘set’, ‘name’, ‘jack’)这个脚本，语法如下：

EVAL "return redis.call('set', 'name', 'jack')" 0 #调用脚本

脚本本质是字符串，所以用双引号括起来，代表是脚本的内容即script，后面的0是脚本需要的key类型的参数的个数，即脚本需要的key类型的参数个数即numkeys

如果脚本中的key、value不想写死，可以作为参数传递key类型参数会放入KEYS数组，其他参数会放入ARGV数组，在脚本中可以从KEYS和ARGV数组获取这些参数：

#调用脚本(lua语言里数组的下标是从1开始)
EVAL "return redis.call('set', KEYS[1], ARGV[1])" 1 name Rose

用Lua脚本编写释放锁的业务流程

-- 锁的key，这是key类型的参数，将来放在KEY数组
-- local key = KEYS[1]
-- 当前线程标示
-- local thredId = ARGV[1]-- 获取锁中的线程标示，get key
local id = redis.call('get', KEYS[1])-- 判断是否与指定的标示（当前线程标示）一致
if(id == ARGV[1]) then-- 如果一致释放锁 del keyreturn redis.call('del', KEYS[1])
end
return 0 -- 不一致就返回0

基于lua脚本修改Redis的分布式锁的释放逻辑

代码示例3：修改锁的实现类

SimpleRedisLock.java

public class SimpleRedisLock implements ILock {public String name;// 业务名，不同的业务的锁的key要不同（实际上就是key名）。private StringRedisTemplate stringRedisTemplate;// 通过构造函数传值public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {this.name = name;this.stringRedisTemplate = stringRedisTemplate;}private static final String KEY_PREFIX = "lock:";//key的前缀private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";// 使用hutool工具类的，传true就能去掉横线，比较方便private static final DefultRedisScript<Long> UNLOCK_SCRIPT;static {// 随着类的加载而执行，并且只会执行一次，因为这玩意（unlock.lua）加载一次可以，没必要每次都加载UNLOCK_SCRIPT = new DefaultRedisScript<>();UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));UNLOCK_SCRIPT.setResultType(Long.class);// 设置返回值为long}@Overridepublic boolean tryLock(long timeoutSec) {// 获取当前线程的id（线程的标识），前面拼接UUID，保证集群下的唯一String threadId = ID_PREFIX + Thread.currentThread().getId();// 获取锁，setIfAbsent 如果不存在则执行。value最好加个标识，哪个线程拿到的锁。Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);// 返回值是boolean基本类型，而这里是Boolean，所以直接返回的话会自动拆箱，如果值是null，拆箱时可能会报空指针错误，所以返回时可以这么返回，防止自动拆箱return Boolean.TURE.equals(success);// Boolean.TURE是常亮，所以与success比较后，一样就返回true，不一样（或是null时）就返回false。}/*@Overridepublic void unlock() {// 获取线程标示String threadId = ID_PREFIX + Thread.currentThread().getId();// 获取锁中的标示String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);// 就是key的值// 判断标示是否一致if (threadId.equals(id)) {// 一致时才删stringRedisTemplate.delete(KEY_PREFIX + name);}}*/@Overridepublic void unlock() {// 调用lua脚本，用excute函数// 参数1：script 参数2：keys（list类型）stringRedisTemplate.excute(UNLOCK_SCRIPT,Collections.singletonList(KEY_PREFIX + name),// 单元素的listID_PREFIX + Thread.currentThread().getId());}
}

新建lua script 文件 /java/resources/unlock.lua

-- 获取锁中的线程标示，get key
local id = redis.call('get', KEYS[1])-- 判断是否与指定的标示（当前线程标示）一致
if(id == ARGV[1]) then-- 如果一致释放锁 del keyreturn redis.call('del', KEYS[1])
end
return 0 -- 不一致就返回0

这样的话，基本上是生产可用的相对完善的分布式锁。

Redis 基础 - 优惠券秒杀《分布式锁（初级）》相关推荐

Redis 基础 - 优惠券秒杀《初步优化（异步秒杀）》
Redis基础 - 基本类型及常用命令 Redis基础 - Java客户端 Redis 基础 - 短信验证码登录 Redis 基础 - 用Redis查询商户信息 Redis 基础 - 优惠券秒杀< ...
Redis 基础 - 优惠券秒杀《非集群》
参考 Redis基础 - 基本类型及常用命令 Redis基础 - Java客户端 Redis 基础 - 短信验证码登录 Redis 基础 - 用Redis查询商户信息摘要用Redis生成保证唯一性 ...
Redis基础与高可用集群架构进阶详解
一.NoSQL简介 1.问题引入每年到了过年期间,大家都会自觉自发的组织一场活动,叫做春运!以前我们买票都是到火车站排队,后来呢,有了 12306,有了它以后就更方便了,我们可以在网上买票,但是带来 ...
Redis基础（八）——集群
文章目录集群 1 主从复制 1.1 建立连接阶段 1.2 数据同步阶段 1.3 命令传播阶段 2 哨兵 2.1 监控阶段 2.2 通知阶段 2.3 故障转移阶段 3 集群集群 1 主从复制为提高 ...
Redis进阶-高可用：集群
前言前面几篇文章中,已经介绍了Redis的几种高可用技术:持久化.主从复制和哨兵,但这些方案仍有不足,其中最主要的问题是存储能力受单机限制,以及无法实现写操作的负载均衡. Redis集群解决了上述 ...
SpringBoot使用Redis 数据访问(单点、集群、哨兵、连接池、Pipline、分布式框架Redisson、解决方案)
目录 Redis 文献资料用Redis编程 Redis模块API 教程和常见问题解答管理嵌入式和物联网故障排除 Redis集群其他基于Redis的分布式系统在SSD和永久性存储器上进行Re ...
Redis(3)--哨兵模式，集群
Redis的哨兵模式什么是哨兵模式 Redis提供了哨兵的命令,哨兵是一个独立的进程,作为进程,它会独立运行.其原理是哨兵通过发送命令,等待Redis服务器响应,从而监控运行的多个Redis实例. ...
NoSQL(3) 之Redis主从复制、哨兵和集群介绍及详细搭建步骤
文章目录一.主从复制 1.1 主从复制的概念 1.2 主从复制的作用 1.3 主从复制的流程 1.4 部署Redis 主从复制步骤 1)首先要搭建redis,在之前的博客写过, 具体可参考:NoSQ ...
Redis系列教程(二)：详解Redis的存储类型、集群架构、以及应用场景
高并发架构系列高并发架构系列:数据库主从同步的3种一致性方案实现,及优劣比较高并发架构系列:Spring Cloud的核心成员.以及架构实现详细介绍高并发架构系列:服务注册与发现的实现原理.及实 ...

Redis 基础 - 优惠券秒杀《分布式锁（初级）》

参考

synchronized在集群上使用时的问题

为何需要分布式锁

业务场景

分布式锁的实现

基于Redis的分布式锁

基于Redis实现分布式锁（初级）

代码示例1：用一人一单的例子来测试

Redis分布式锁误删问题

解决误删问题的办法

代码示例2：修改锁的实现类

原子性问题

lua脚本解决多条命令原子性问题

用Lua脚本编写释放锁的业务流程

基于lua脚本修改Redis的分布式锁的释放逻辑

代码示例3：修改锁的实现类

Redis 基础 - 优惠券秒杀《分布式锁（初级）》相关推荐

最新文章

热门文章