解密高并发系统实战内幕

Posted 2022-12-06 愉悦滴帮主)

前言：

火爆的双十一活动大家都了解，其中的秒杀商品更是让我又爱又恨。但是作为一个开发者来说，双十一中所涉及的高并发问题也是很头痛的。一些大厂像京东，阿里的面试中，通常都会问道高并发的问题。其中以 “如何设计一个秒杀系统”、“微信抢红包” 这两个场景最为经典。

---

秒杀系统核心业务架构设计剖析

业务场景分析：

 首先我们先看下秒杀场景的问题都在哪？在秒杀场景中容易产生大量的并发请求，从而产生库存扣减问题。分别是：超卖，少卖。

1）超卖：超卖的场景就是说，本来库存中有一百件商品，结果产生了一百多个订单。

2）少卖：少卖的场景就是说，本来库存中有一百件商品，发现卖出去99件之后，库存服务告诉你商品卖完了。

秒杀除了大并发这样的难点，超卖，少卖电商都会遇到的痛，电商搞大促最怕什么？最怕的就是超卖，少卖。产生上述问题以后会直接影响到用户体验，会导致订单系统、库存系统、供应链等等，产生的问题是一系列的连锁反应，所以电商都不希望这样的问题发生，但是在大并发的场景最容易发生的就是超卖，少卖，不同线程读取到的当前库存数据可能下个毫秒就被其他线程修改了，如果没有一定的锁库存机制那么库存数据必然出错，都不用上万并发，几十并发就可以导致商品超卖或少卖；

---

实操：从手把手写代码实现高并发库存扣减

我们通过下述代码模拟电商系统的库存服务。下面代码的主要意思就是：首先去redis缓存中查剩余库存数，如果大于零件，就库存数减一，刷新缓存的一个操作。

package com.tguo.demo.service;

import com.tguo.demo.service.impl.StockServiceImpl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

@Service
public class StockService implements StockServiceImpl 
    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public String reduce() 
        int stock;
        if((stock= (int) redisTemplate.opsForValue().get("stock"))>0)
            int remainStock = stock -1; //业务逻辑
            redisTemplate.opsForValue().set("stock",remainStock); //库存更新

            System.out.println("秒杀成功，当前剩余库存："+remainStock);
            return  "success";
        
        System.out.println("秒杀失败，当前剩余库存："+stock);
        return "fail";
    

我们通过压测工具jmeter测试，同时并发一百条请求后发现打印结果如下。通过下面结果分析：我们发生了上述所说超卖的问题。

秒杀成功，当前剩余库存：17
秒杀成功，当前剩余库存：18
秒杀成功，当前剩余库存：16
秒杀成功，当前剩余库存：16
秒杀成功，当前剩余库存：15
秒杀成功，当前剩余库存：14
秒杀成功，当前剩余库存：13
秒杀成功，当前剩余库存：13
秒杀成功，当前剩余库存：12
秒杀成功，当前剩余库存：12
秒杀成功，当前剩余库存：12
秒杀成功，当前剩余库存：11
秒杀成功，当前剩余库存：11
秒杀成功，当前剩余库存：10

 我们想到的解决办法就是sync加锁。这样做在一体机式项目可以解决该问题。但是在淘宝双十一的项目基本都是微服务项目，而且为了分担单个服务压力，比如库存服务，会将库存服务做成集群的形式。如果库存服务为集群。我们通过sync加锁就不能解决上述超卖问题。

因为synchronized关键字的作用域其实是一个进程，在这个进程下面的所有线程都能够进行加锁。但是多进程就不行了。对于秒杀商品来说，这个值是固定的。但是每个地区都可能有一台服务器。这样不同地区服务器不一样，地址不一样，进程也不一样。因此synchronized无法保证数据的一致性。synchronized与lock锁都是作用于同一进程里面，因为多个线程共同访问某个共享资源，而进行的同步措施，他的前提条件是同一进程内，内存共享；

---

这时候就需要分布式锁。

redis锁底层实现原理以及应用

1. 使用Redis的 SETNX 命令可以实现分布式锁

命令格式
SETNX key value

将 key 的值设为 value，当且仅当 key 不存在。
若给定的 key 已经存在，则 SETNX 不做任何动作。
SETNX 是SET if Not eXists的简写。

返回值
返回整数，具体为
- 1，当 key 的值被设置
- 0，当 key 的值没被设置

因为redis式单线程的，当大量并发请求去请求后台时，后台作集群分摊个服务压力，后统一去调用redis去访问数据。这样便做到了资源共享，资源统一。

@Service
public class StockService implements StockServiceImpl 

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public String reduce() 
        int stock;
        //分布式锁
        String result = "fail";
        //判断是否有锁   setnx stock 1 如果key存在返回0，不能设置成功
        //判断是否有其他线程在调用
        if(!redisTemplate.opsForValue().setIfAbsent(RedisKey.STOCK_MUTE_LOCK,1))
            return  result;
        
        if((stock= (int) redisTemplate.opsForValue().get("stock"))>0)
            int remainStock = stock -1;
            redisTemplate.opsForValue().set("stock",remainStock);
            System.out.println("秒杀成功，当前剩余库存："+remainStock);
            //解锁 删除对应的key
            redisTemplate.delete(RedisKey.STOCK_MUTE_LOCK);
            return  "success";
        
        System.out.println("秒杀失败，当前剩余库存："+stock);
        return "fail";
    

上述代码,首先通过setnx先去获取锁，如果这时有其他线程在调用，则获取失败。当其他线程让出锁的时候其他线程才能继续获取锁。

以上我们解决了锁的问题来保证原子性，一致性。但是如果我们的业务代码中，出现运行时异常，导致该线程的锁并没有解锁。就会导致其他线程获取不到锁。就是死锁问题。这个问题该如何解决？这时候我们可能会用tryfinally来保证。例如：下面代码

  int stock;
        //分布式锁
        String result = "fail";
        //判断是否有锁   setnx stock 1 如果key存在返回0，不能设置成功
        //判断是否有其他线程在调用
        if(!redisTemplate.opsForValue().setIfAbsent(RedisKey.STOCK_MUTE_LOCK,1))
            return  result;
        
        try
            if((stock= (int) redisTemplate.opsForValue().get("stock"))>0)
                int remainStock = stock -1;
                redisTemplate.opsForValue().set("stock",remainStock);
                System.out.println("秒杀成功，当前剩余库存："+remainStock);
                //解锁 删除对应的key
                return  "success";
            
        finally 
            redisTemplate.delete(RedisKey.STOCK_MUTE_LOCK);

        
        System.out.println("秒杀失败，当前剩余库存："+stock);
        return "fail";

用finally来保证死锁问题，这样做确实可以解决问题，但是这只是最后的手段。比如：你业务代码太多而且比较分散的时候，只能try部分代码但是其他代码巡行时抛出异常该怎么办。

解决办法：对锁设置过期时间 redisTemplate.expire

Boolean ifAbsent = redisTemplate.opsForValue().setIfAbsent(RedisKey.STOCK_MUTE_LOCK, 1);
redisTemplate.expire(RedisKey.STOCK_MUTE_LOCK, 3,TimeUnit.SECONDS);

我们通过对锁设置过期时间，来防止即使tryfinally后，其他代码出现异常而导致死锁问题的出现。但是如果

Boolean ifAbsent = redisTemplate.opsForValue().setIfAbsent(RedisKey.STOCK_MUTE_LOCK, 1);

在运行上述代码的时候JVM崩溃了，宕机了导致后续代码没执行。过期时间设置未生效。就很尴尬。 我们需要将上述两步操作改成一步完成，来保证原子性。当然还有其他解决办法，如redis的事务【redis.exec】将上述两步骤放到一个事务中等等。

@Service
public class StockService implements StockServiceImpl 

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public String reduce() 
        int stock;
        //分布式锁
        String result = "fail";
        //判断是否有锁,如果有锁表示其他线程在占用，如果没有表示可以获取锁。并设置过期时间保证原子性
        Boolean ifAbsent = redisTemplate.opsForValue().setIfAbsent(RedisKey.STOCK_MUTE_LOCK, 1, 3, TimeUnit.SECONDS);
        if (!ifAbsent) 
            return result;
        
        try 
            if ((stock = (int) redisTemplate.opsForValue().get("stock")) > 0) 
                //业务代码
                int remainStock = stock - 1;
                redisTemplate.opsForValue().set("stock", remainStock);
                System.out.println("秒杀成功，当前剩余库存：" + remainStock);
                //解锁 删除对应的key
                return "success";
            
         finally 
            //防止死锁问题发生
            redisTemplate.delete(RedisKey.STOCK_MUTE_LOCK);

        
        System.out.println("秒杀失败，当前剩余库存：" + stock);
        return "fail";
    

这时候以目前的代码来说，在高并发场景下大概率没什么问题。算是过关。 

---

lua脚本究竟是何方神圣

上述代码通过分布式锁与设置过期时间简单的保证了并发安全。但是还存在问题，我们分析下图。

问题1：解锁的时候身份不符的问题？

上图中，我们进行分析。首先线程一获取锁，过了5秒的时候锁过期了，但是线程一还没跑完。这时候线程2进来了。又过了两秒后线程一执行完毕后进行解锁，很有可能导致把线程二的锁进行解除。所以我们需要对线程进行身份标识，来防止上述问题的出现。

注意：身份标识必须作为全局唯一ID。可以用用户tocken。

例如：

@Service
public class StockService implements StockServiceImpl 

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public String reduce() 
        int stock;
        //分布式锁
        String result = "fail";
        //身份标识 全局唯一ID  这里用随机数代替。项目上可以用用户tocken。
        String tokenId = UUID.randomUUID().toString();
        //判断是否有锁,如果有锁表示其他线程在占用，如果没有表示可以获取锁。并设置过期时间保证原子性
        Boolean ifAbsent = redisTemplate.opsForValue().setIfAbsent(RedisKey.STOCK_MUTE_LOCK, tokenId, 3, TimeUnit.SECONDS);
        if (!ifAbsent) 
            return result;
        
        try 
            if ((stock = (int) redisTemplate.opsForValue().get("stock")) > 0) 
                //业务代码
                int remainStock = stock - 1;
                redisTemplate.opsForValue().set("stock", remainStock);
                System.out.println("秒杀成功，当前剩余库存：" + remainStock);
                //解锁 删除对应的key
                return "success";
            
         finally 
            //解锁的时候判断是否与加锁对象一致
            if(redisTemplate.opsForValue().get(RedisKey.STOCK_MUTE_LOCK).equals(tokenId))
                redisTemplate.delete(RedisKey.STOCK_MUTE_LOCK);  
            
           

        
        System.out.println("秒杀失败，当前剩余库存：" + stock);
        return "fail";
    

---

问题2：锁的周期续约问题。

但是在项目线上环境我们是没有有效的手段来测出锁的时间的。锁的有效时间只能凭借开发经验来设定。这就会因为一些硬件，网络或者其他的一些因素导致线程还没执行完毕后锁失效了。针对上述问题我们期望，如果锁还存在则延长锁的生命周期，解决线程还未完成的时候锁失效问题。

如何解决：要求（1.保证原子性。2.自动续约所得生命周期。3.解决对象与加锁对象一致）

这里我们通过lua脚本来解决上面的问题。

//加锁    
public List luaLock(String key)
    	String id = UUID.randomUUID().toString();
    	String script = "if (redis.call('exists',KEYS[1])==0) then"+   //判断锁是不是存在
				"redis.call('hincrby',KEYS[1],ARGV[2],1);"+            //不存在则加锁
				"redis.call('pexpire',KEYS[1],ARGV[1]);"+              //设置过期时间
				"return nil;"+
				"end;"+
				"if(redis.call('hexists',KEYS[1],ARGV[2])==1) then"+  //如果存在，
				"redis.call('hincrby',KEYS[1],ARGV[2],1);"+            //锁+1 = 重入锁
    	        "redis.call('pexpire',KEYS[1],ARGV[1]);"+              //再设置过期时间
    	        "return nil;"+
				"end;"+
				"return redis.call('pttl',KEYS[1])";                   //返回ttl
		DefaultRedisScript<List> redisScript = new DefaultRedisScript<>(script);
		redisScript.setResultType(List.class);
		return (List)redisTemplate.execute(redisScript, Arrays.asList(key),30000,id);
	

/**
	 * 解锁
	 * @param key
	 * @return
	 */
	public List luaUnlock(String key)
		String id = UUID.randomUUID().toString();
		String script = "if (redis.call('hexists',KEYS[1],ARGV[3])==0) then"+
				"return nil;"+
				"end;"+
				"local counter = redis.call('hincry',KEYS[1],ARGV[3],-1);"+  //重入锁 -1
				"if(counter>0);"+
				"else"+
				"redis.call('del',KEYS[1]);"+    //重入锁=0 解锁
				"redis.call('publish',KEYS[2],ARGV[1]);"+
				"return 1;"+
				"end;"+
				"return nil";
		DefaultRedisScript<List> redisScript = new DefaultRedisScript<>(script);
		redisScript.setResultType(List.class);
		return (List)redisTemplate.execute(redisScript, Arrays.asList(key),30000,id);
	

 多锁情况: muliple lock。参考：https://blog.csdn.net/weixin_39800144/article/details/84624637

redisson中的MultiLock，可以把一组锁当作一个锁来加锁和释放。基于Redis的分布式RedissonMultiLock对象将多个RLock对象分组，并将它们作为一个锁处理。每个RLock对象可能属于不同的Redisson实例。

RLock lock1 = redissonInstance1.getLock("lock1");
RLock lock2 = redissonInstance2.getLock("lock2");
RLock lock3 = redissonInstance3.getLock("lock3");

RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
// locks: lock1 lock2 lock3
lock.lock();
...
lock.unlock();

以上就完成了分布式锁的实现

---

基于Redission实现分布式锁

 redssion锁的原理这里就不叙述了。

想了解的小伙伴请参考：https://www.jianshu.com/p/67f700fad8b3

---

 

剖析Zookeeper锁原理

首先再并发场景下客户端请求过来会进行竞争，竞争到的会与Zookeeper建立临时节点。当这个请求执行完毕后，会删除该节点，这时会被Zookeeper监听，并产生一个时间通知其他请求。表示欢迎下一位进场。

项目资源：https://download.csdn.net/download/qq_45065241/19731396