随着互联网的高速发展，商品秒杀场景并不少见；秒杀是一个紧缺的高并发场景，包含很多技术点，掌握了技术点，不一定能让你马上面试成功，但一定是一个闪光点！前言假设我们现在有一个商城系统，有一个产品在线闪购模块，那么我们如何设计这个模块呢？秒杀模块有哪些不同的要求？全球唯一ID商品秒杀其实就是商品购买，所以我们需要准备一个订单表来记录对应的秒杀订单。这就涉及到一个orderid的问题。能不能像其他表一样使用数据库自??带的自增??id呢？数据库自增id的缺点如果订单表使用数据库自??增id，会出现一些问题：id的规律太明显了，因为我们的订单id需要回显给用户查看，如果id太明显，会暴露一些信息，比如第一天的订单id=10，第二天的订单id=11，也就是说中间没有其他用户下单两个命令。单表的数据量是有限的。在高并发场景下，有可能产生百万级的订单，我们都知道MySQL单张表是不可能放这么多数据的（受限于性能等原因）；如果将单表拆成多表，或者使用数据库本身添加id的话，订单id就会重复，这显然是业务不允许的。基于以上两个问题，我们可以知道订单表的id需要是一个全局唯一的ID，不能有明显的规则。全局ID生成器全局ID生成器是用于在分布式系统中生成全局唯一ID的工具。它一般满足以下特点：这里我们思考Redis中的自增计数是否可以作为一个全局id生成器呢？能不能主要看是否满足以上五个条件：唯一性，每个订单都来Redis生成订单id，所以唯一性才能保证高可用，Redis通过主从、集群、集群等方式保证高可用和高性能其他模式，Redis是基于内存的，在性能上本质上是增量的，在安全上增量本质上是增量的。..这个比较麻烦一点，因为Redis的增量也在递增，这个规律太明显了。..综上所述，Redis的增量并不能满足安全性，所以我们不能简单的将其作为一个全局的id生成器。但是-我们可以用它和其他东西拼接~举个栗子：ID的组成部分：signbit:1bit,always0Timestamp:31bit,inseconds,canuse69yearsSerialnumber:32bit,一个秒内的计数器，支持每秒生成2^32个不同的ID。上面的时间戳是用来增加复杂度的。下面给出了以下代码示例：publicclassRedisIdWorker{/***starttimestamp*/privatestaticfinallongBEGIN_TIMESTAMP=1640995200L;/***序列号的位数*/privatestaticfinalintCOUNT_BITS=32;privateStringRedisTemplatestringRedisTemplate;publicRedisIdWorker(StringRedisTemplatestringRedisTemplate){this.stringRedisTemplate=stringRedisTemplate;}publiclongnextId(StringkeyPrefix){//1.生成时间戳LocalDateTimenow=LocalDateTime.now();longnowSecond=now.toEpochSecond(ZoneOffset.UTC);长时间戳=nowSecond-BEGIN_TIMESTAMP；获取当前日期，精确到天Stringdate=now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));//2.2。自增长//一天一个keylongcount=stringRedisTemplate.opsForValue().increment("icr:"+keyPrefix+":"+date);//3.拼接返回returntimestamp<0.update();请注意，上面的CAS判断做了优化，不是判断刚刚查询到的库存是否等于扣减时的库存，而是判断当前库存是否大于0。因为会出现问题判断刚刚查询到的存货与扣款时的存货是否相等：如果多个线程判断不相等，则全部停止扣款，这时候就没有办法全部购买了。而判断当前库存是否大于0就可以很好的解决上面的问题！一人一单的需求一般来说，闪购的商品都是大打折扣的，所以可能会有一个需求——平台只允许一个用户购买一件商品。在秒杀场景下，我们应该如何针对这个需求进行设计呢？显然，在执行扣除库存的操作之前，我们需要检查数据库是否已经有用户的订单；如果是，说明用户已经下单，不能再购买；如果不是，则执行Deduct操作并生成订单。//查询订单intcount=query().eq("user_id",userId).eq("voucher_id",voucherId).count();//判断是否存在if(count>0){//用户hasalreadypurchaseAfterreturnResult.fail("用户已经购买过一次！");}//扣除库存booleansuccess=seckillVoucherService.update().setSql("stock=stock-1")//setstock=stock-1.eq("voucher_id",voucherId).gt("stock",0)//其中id=?和库存>0.update();并发安全问题因为上面的实现分为两步：判断当前用户数据库中没有订单进行扣款操作，订单的生成分为两步，导致线程安全问题：可以是同一个用户的多个请求线程同时判断没有订单，然后每个人都进行了扣款操作。解决这个问题也很简单，只要让这两个步骤串行执行即可，即加锁！在方法头中加入synchronized显然会锁住整个方法，锁的范围太大，会限制所有的请求线程；而我们的需求只是同一用户的串行请求线程；明显有些大材小用了~@TransactionalpublicsynchronizedResultcreateVoucherOrder(LongvoucherId){//一人一单LonguserId=UserHolder.getUser().getId//查询订单intcount=query().eq("user_id",userId).eq("voucher_id",voucherId).count();//判断是否存在if(count>0){//用户已经购买过returnResult.fail("用户已经购买过一次！");//扣除stockbooleansuccess=seckillVoucherService.update().setSql("stock=stock-1")//setstock=stock-1.eq("voucher_id",voucherId).gt("stock",0)//其中ID=？和库存>0.update();if(!success){//扣费失败returnResult.fail("存货不足！");//创建订单VoucherOrdervoucherOrder=newVoucherOrder();.....returnResult.ok(orderId);}锁定相同用户id的StringObject@TransactionalpublicResultcreateVoucherOrder(LongvoucherId){//一人一单LonguserId=UserHolder.getUser().getId//锁定相同用户id的String对象同步(userId.toString().intern()){//查询订单intcount=query().eq("user_id",userId).eq("voucher_id",voucherId).count();//判断是否存在......//扣除库存......//创建订单......}}returnResult.ok(orderId);}上面的方法启动了事务，但是是同步的(userId.toString().intern())没有对整个方法加锁（先释放锁，再提交事务，写订单），那么就有问题了——如果某个线程的事务还没有提交（也就是订单还没有写），这时候其他线程来了但是可以拿到锁，它判断数据库中的订单为0，可以重新创建订单。..为了解决这个问题，我们需要先提交事务，然后释放锁：//LocktheStringobjectofthesameuseridsynchronized(userId.toString().intern()){...createVoucherOrder(voucherId);......}@TransactionalpublicResultcreateVoucherOrder(LongvoucherId){//一人一单LonguserId=UserHolder.getUser().getId//查询订单intcount=query().eq("user_id",userId).eq("voucher_id",voucherId).count();//判断是否存在......//扣除库存...//创建订单...returnResult.ok(orderId);}集群模式下的并发安全问题刚才讲的都是stand-默认情况下是单独的节点，但是现在如果将它们放在集群模式下，就会出现一些问题。刚才的加锁解决了单机节点下的线程安全问题，但是无法解决集群下多节点的线程安全问题：因为synchronized锁对应的是JVM中的锁监视器，但是不同的节点有不同的JVM，不同的JVM有不同的锁监视器，所以刚才的设计以集群方式锁不同的对象，也就是解决不了线程安全问题。知道了问题的原因，我们应该很快想出解决办法：既然因为集群的原因，锁不一样，那我们重新设计一下，让它们都使用同一个锁吧！Distributedlocks分布式锁：满足分布式系统或集群模式下多进程可见互斥锁的要求。分布式锁的实现分布式锁的核心是实现多个进程之间的互斥，而满足这一点的方式有很多种，常见的有以下三种：排斥如setnx命令利用节点的唯一性和顺序来实现互斥。高可用性很好。高性能通常是好的。一般安全。断开和自动释放锁使用锁超时释放临时节点，超时自动释放基于Redis的分布式锁使用Redis实现分布式锁，主要应用于SETNX键值命令（如果不存在，设置它）实现两个功能：获取锁（设置一个key）释放锁（删除一个key）基本思路是执行SETNX命令的线程获取锁。完成操作后，需要删除key并释放锁。Locking:@OverridepublicbooleantryLock(longtimeoutSec){//获取线程IDStringthreadId=ID_PREFIX+Thread.currentThread().getId();//获取锁Booleansuccess=stringRedisTemplate.opsForValue()IXname.setIfAbsent(+,threadId,timeoutSec,TimeUnit.SECONDS);returnBoolean.TRUE.equals(success);}释放锁：@Overridepublicvoidunlock(){//获取线程IDStringthreadId=ID_PREFIX+Thread.currentThread().getId();//获取锁中的标记Stringid=stringRedisTemplate.opsForValue().get(KEY_PREFIX+name);//释放锁stringRedisTemplate.delete(KEY_PREFIX+name);}但是这里会有一个隐患——假设线程发生阻塞（或者其他问题），锁还没有释放怎么办（删除key)?为了解决这个问题，我们需要给key设计一个超时时间，让它过期；但是超时时间的长短不好确定：如果设置的太短，会导致其他线程提前获取到锁，造成线程安全问题。对于误删除需要等待额外锁的线程来说，超时时间是很难掌握的，因为业务线程的阻塞时间是不可预测的。极端情况下，可以一直阻塞，直到锁超时超时，如上图线程1，锁超时被释放，导致线程2也进来，此时锁是线程2的锁（key相同，value不同，value一般为线程的唯一标识）；假设此时，线程1突然不阻塞了，它要释放锁。按照刚才的代码逻辑，会释放线程2的锁；线程2的锁释放后，会导致其他线程进来（线程3），以此类推。..为了解决这个问题，需要在释放锁的时候额外增加一个判断。每个线程只释放自己的锁，不能释放别人的锁！释放锁@Overridepublicvoidunlock(){//获取线程IDStringthreadId=ID_PREFIX+Thread.currentThread().getId();//获取锁中的IDStringid=stringRedisTemplate.opsForValue().get(KEY_PREFIX+name);//判断标记是否一致if(threadId.equals(id)){//释放锁stringRedisTemplate.delete(KEY_PREFIX+name);线程锁当前线程释放锁。可以看到释放锁分两步完成。如果你对并发比较了解，应该马上就知道这里会出问题。步步为营，并发问题！假设线程1判断当前锁是自己的锁，准备释放锁，但是此时阻塞了（可能是FULLGC造成的），锁超时，线程2加锁。这时候，锁是线程2的；但是如果此时线程1被唤醒，因为它已经执行完了step1，所以这时候会直接去step2去释放锁（但是此时的锁不是thread1的）其实这是a原子性的问题，刚刚释放锁的那两步应该是原子的，不可分割的！要使其具有原子性，您需要在Redis中使用Lua脚本。引入Lua脚本保持原子性Lua脚本：--比较线程ID和锁IDif(redis.call('get',KEYS[1])==ARGV[1])then--释放锁delkeyreturnredis.call('del',KEYS[1])endreturn0Java中的调用执行：publicclassSimpleRedisLockimplementsILock{privateStringname;privateStringRedisTemplatestringRedisTemplate;publicSimpleRedisLock(Stringname,StringRedisTemplatename.stringthis){RedisTemplate)name;this.stringRedisTemplate=stringRedisTemplate;}privatestaticfinalStringKEY_PREFIX="lock:";privatestaticfinalStringID_PREFIX=UUID.randomUUID().toString(true)+"-";CKprivatestaticfinalSCRIPTScri;static{UNLOCK_SCRIPT=newDefaultRedisScript<>();UNLOCK_SCRIPT.setLocation(newClassPathResource("unlock.lua"));UNLOCK_SCRIPT.setResultType(Long.class);}@OverridepublicbooleantryLock(longtimeoutSec){//获取线程标记StringthreadId=ID_PREFIX+Thread.currentThread().getId();//获取锁Booleansuccess=stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX+name,threadId,timeoutSec,TimeUnit.SECONDS);返回Boolean.TRUE.equals（成功）；}@Overridepublicvoidunlock（）{//lualuastringRedistemplate.execute（unlock_script，collections.singletonlist（key_prefix+name），id_prefix+thread.current.currentthreadReadRead（）。}}至此，我们设计的Redis分布式锁已经是比较完善的可用于生产的分布式锁总结本次我们从秒杀场景的业务需求出发，一步步使用Redis设计了一个生产可用的分布式锁：实现思路：使用setnxex获取锁，并设置过期时间，保存threadmark和先释放锁判断threadmark和自己是否一致，如果一致则删除锁有什么特点（Lua脚本保证原子性）？使用setnx满足互斥使用setex保证故障时仍然可以释放锁，避免死锁，提高安全性使用Redis集群保证高可用和高并发特性还有待提高：不可重入，同一个线程不能多次获取同一个锁不能重试。仅获取一次锁将返回false。没有超时释放的重试机制。锁超时释放虽然可以避免死锁，但是如果业务执行耗时较长，也会导致锁被释放，存在安全隐患（虽然解决了误删问题，但可能仍然存在未知问题）主从一致性，如果Redis提供主从集群，主从同步存在延迟，当主下来，主节点中的锁如果没有及时将数据同步到从节点，其他线程也会获取到锁，造成线程安全问题（延迟时间在毫秒以下，所以这个概率情况极低）
