Spring Boot整合Redis实现简单的分布式锁

电玩女神 2022-01-23 23:27 444阅读 0赞

随着微服务等分布式架构的快速发展及应用，在很多情况下，我们都会遇到在并发情况下多个线程竞争资源的情况，比如我们耳熟能详的秒杀活动，多平台多用户对同一个资源进行操作等场景等。分布式锁的实现方式有很多种，比如基于数据库、Zookeeper、Redis等，本文我们主要介绍Spring Boot整合Redis实现分布式锁。

**1、为什么需要分布式锁？**

在开发应用时，当多个客户或者多个线程需要对某个共享的数据进行操作时，就需要使用线程同步。在Java开发中，对于单机应用，因为是在同一个JVM内部，所以我们可以采用Java提供的各种多线程操作的技巧来实现线程同步。

而对于分布式系统来说，由于多个请求可能被分发到不同的机器上去处理，如果这多个请求都是对同一个资源进行操作，那么使用基本的Java多线程线程同步技术可能就解决不了这个问题。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5aGx3enk_size_16_color_FFFFFF_t_70]  
如上图，请求A、B、C都是发起扣减同一个商品的库存操作，三个请求被分发到三台不同的服务部署机器上进行处理。而三台机器并不在同一个JVM，所以Java提供的线程同步技巧就发挥不了作用了。但是对于扣减库存这样的场景，必须要使用线程同步来保证同一个商品的库存不会被漏扣或者多扣。

为了保证在高并发的场景下，临界资源（共享资源）同时只能被一个线程执行，在传统单体应用单机部署的情况下，可以使用Java并发处理相关的API(如ReentrantLock或Synchronized)进行互斥控制。在单机环境中，Java中提供了很多并发处理相关的API。

但是在分布式系统中，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题！

**2、分布式锁应该具备哪些条件？**

*  在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行；
 *  高可用的获取锁与释放锁；
 *  高性能的获取锁与释放锁；
 *  具备可重入特性；
 *  具备锁失效机制，防止死锁；
 *  具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败。

**3、Spring Boot整合Redis实现简单的分布式锁**

*实现思想*

1.  SETNX key val：当且仅当key不存在时，set一个key为val的字符串，返回1；若key存在，则什么都不做，返回0。
2.  expire key timeout：为key设置一个超时时间，单位为second，超过这个时间锁会自动释放，避免死锁。
3.  delete key：删除key

*Spring Boot整合Redis实现简单的分布式锁实现细节*

我们以实现一个多个线程同步扣除同一个商品的库存为例，实现一个简单的Redis分布式锁。实例需要依赖的内容如下：

1.  Spring Boot Web 依赖：通过在页面上点击实现多个请求；
2.  Spring JPA：数据库访问；
3.  MySQL：存储商品库存；
4.  Thymeleaf：页面模板；

按照如下的项目结构创建项目。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5aGx3enk_size_16_color_FFFFFF_t_70 1]

pom.xml如下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    	<modelVersion>4.0.0</modelVersion>
    	<parent>
    		<groupId>org.springframework.boot</groupId>
    		<artifactId>spring-boot-starter-parent</artifactId>
    		<version>2.1.5.RELEASE</version>
    		<relativePath/> 
    	</parent>
    	<groupId>com.distributedlock</groupId>
    	<artifactId>redis_dis_lock</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    	<name>redis_dis_lock</name>
    	<description>Redis实现分布式锁</description>
    
    	<properties>
    		<java.version>1.8</java.version>
    	</properties>
    
    	<dependencies>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-data-jpa</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-data-redis</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-web</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-thymeleaf</artifactId>
    		</dependency>
    		<dependency>
    			<groupId>mysql</groupId>
    			<artifactId>mysql-connector-java</artifactId>
    			<scope>runtime</scope>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-test</artifactId>
    			<scope>test</scope>
    		</dependency>
    	</dependencies>
    
    	<build>
    		<plugins>
    			<plugin>
    				<groupId>org.springframework.boot</groupId>
    				<artifactId>spring-boot-maven-plugin</artifactId>
    			</plugin>
    		</plugins>
    	</build>
    
    </project>

依赖配置好之后，如果没有安装Redis和MySQL的，自己先安装好Redis和MySQL，可参考如下两篇文章。

Spring整合Redis实现数据缓存（https://blog.csdn.net/zyhlwzy/article/details/54575386）  
Linux上安装MySQL（https://blog.csdn.net/zyhlwzy/article/details/80569174）

准备工作做好之后，在Spring Boot配置文件中配置Redis、MySQL的链接属性、Spring Boot应用端口、名称等。

spring:
      application:
        name: Redis Distribute Lock
    
      redis:
        host: 127.0.0.1
        port: 6379
        timeout: 20000
        password: 654321
        jedis:
          pool:
            max-active: 8
            min-idle: 0
            max-idle: 8
            max-wait: -1
          
      datasource:
        url: jdbc:mysql://127.0.0.1:3306/test?useUnicode=true&characterEncoding=UTF8&zeroDateTimeBehavior=convertToNull&useSSL=true&allowMultiQueries=true&serverTimezone=Asia/Hong_Kong
        username: root
        password: 111111
        driver-class-name: com.mysql.jdbc.Driver
          
      jpa:
        show-sql: true
        hibernate:
          ddl-auto: none
          
    server:
      port: 8090

定义库存实体

import java.io.Serializable;
    import javax.persistence.Column;
    import javax.persistence.Entity;
    import javax.persistence.Id;
    import javax.persistence.Table;
    
    @Entity
    @Table(name="goods_store")
    public class GoodsStore implements Serializable {
    
    	/**
    	 * 
    	 */
    	private static final long serialVersionUID = 1L;
    	
    	@Id
    	private String code;
    	
    	@Column(name="store")
    	private int store;
    
    	//get、set省略
    }

实现Redis分布式锁

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.stereotype.Component;
    import org.springframework.util.StringUtils;
    
    @Component
    public class RedisLock {
    	@Autowired
    	private StringRedisTemplate stringRedisTemplate;
    	
    	/**
    	 * 加锁
    	 * @param lockKey 加锁的Key
    	 * @param timeStamp 时间戳：当前时间+超时时间
    	 * @return
    	 */
    	public boolean lock(String lockKey,String timeStamp){
    		if(stringRedisTemplate.opsForValue().setIfAbsent(lockKey, timeStamp)){
    			// 对应setnx命令，可以成功设置,也就是key不存在，获得锁成功
    			return true;
    		}
    		
    		//设置失败，获得锁失败
    		// 判断锁超时 - 防止原来的操作异常，没有运行解锁操作 ，防止死锁
    		String currentLock = stringRedisTemplate.opsForValue().get(lockKey);
    		// 如果锁过期 currentLock不为空且小于当前时间
    		if(!StringUtils.isEmpty(currentLock) && Long.parseLong(currentLock) < System.currentTimeMillis()){
    			//如果lockKey对应的锁已经存在，获取上一次设置的时间戳之后并重置lockKey对应的锁的时间戳
    			String preLock = stringRedisTemplate.opsForValue().getAndSet(lockKey, timeStamp);
    			
    			//假设两个线程同时进来这里，因为key被占用了，而且锁过期了。
    			//获取的值currentLock=A(get取的旧的值肯定是一样的),两个线程的timeStamp都是B,key都是K.锁时间已经过期了。
    			//而这里面的getAndSet一次只会一个执行，也就是一个执行之后，上一个的timeStamp已经变成了B。
    			//只有一个线程获取的上一个值会是A，另一个线程拿到的值是B。
    			if(!StringUtils.isEmpty(preLock) && preLock.equals(currentLock)){
    				return true;
    			}
    		}
    		
    		return false;
    	}
    	
    	/**
    	 * 释放锁
    	 * @param lockKey
    	 * @param timeStamp
    	 */
    	public void release(String lockKey,String timeStamp){
    		try {
                String currentValue = stringRedisTemplate.opsForValue().get(lockKey);
                if(!StringUtils.isEmpty(currentValue) && currentValue.equals(timeStamp) ){
                    // 删除锁状态
                    stringRedisTemplate.opsForValue().getOperations().delete(lockKey);
                }
            } catch (Exception e) {
                System.out.println("警报！警报！警报！解锁异常");
            }
    	}
    }

创建库存Respository

import org.springframework.data.jpa.repository.JpaRepository;
    import org.springframework.data.jpa.repository.Modifying;
    import org.springframework.data.jpa.repository.Query;
    import org.springframework.data.repository.query.Param;
    import org.springframework.transaction.annotation.Transactional;
    import com.distributedlock.redis.entity.GoodsStore;
    
    /**
     * 库存Respository
     * @author user
     *
     */
    public interface GoodsStoreRespository extends JpaRepository<GoodsStore,String> {
    	/**
    	 * 更新库存
    	 * @param code
    	 * @param store
    	 * @return
    	 */
    	@Modifying
        @Transactional
        @Query("update GoodsStore gs set gs.store=gs.store-?2 where gs.code=?1")
    	int updateStore(@Param("code") String code,@Param("store")Integer store);
    }

创建库存接口，定义更新库存和获取库存信息的方法。

import com.distributedlock.redis.entity.GoodsStore;
    
    public interface GoodsStoreFacade {
    	/**
    	 * 根据产品编号更新库存
    	 * @param code
    	 * @return
    	 */
    	String updateGoodsStore(String code,int count);
    	
    	/**
    	 * 获取库存对象
    	 * @param code
    	 * @return
    	 */
    	GoodsStore getGoodsStore(String code);
    }

实现更新库存接口

import java.util.Optional;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import com.distributedlock.redis.entity.GoodsStore;
    import com.distributedlock.redis.facade.GoodsStoreFacade;
    import com.distributedlock.redis.lock.RedisLock;
    import com.distributedlock.redis.respository.GoodsStoreRespository;
    
    /**
     * 库存管理服务
     * @author user
     *
     */
    @Service
    public class GoodsStoreService implements GoodsStoreFacade{
    	@Autowired
    	private GoodsStoreRespository goodsStoreRespository;
    	
    	@Autowired
    	private RedisLock redisLock;
    	
    	/**
         * 超时时间 5s
         */
    	private static final int TIMEOUT = 5*1000;
    	
    	/**
    	 * 根据产品编号更新库存
    	 * @param code
    	 * @return
    	 */
    	@Override
    	public String updateGoodsStore(String code,int count) {
    		//上锁
    		long time = System.currentTimeMillis() + TIMEOUT;
    		if(!redisLock.lock(code, String.valueOf(time))){
    			return "排队人数太多，请稍后再试.";
    		}
    		System.out.println("获得锁的时间戳："+String.valueOf(time));
    		try {
    			GoodsStore goodsStore = getGoodsStore(code);
    			if(goodsStore != null){
    				if(goodsStore.getStore() <= 0){
    					return "对不起，卖完了，库存为："+goodsStore.getStore();
    				}
    				if(goodsStore.getStore() < count){
    					return "对不起，库存不足，库存为："+goodsStore.getStore()+" 您的购买数量为："+count;
    				}
    				System.out.println("剩余库存："+goodsStore.getStore());
    				System.out.println("扣除库存："+count);
    				goodsStoreRespository.updateStore(code, count);
    				try{
    					//为了更好的测试多线程同时进行库存扣减，在进行数据更新之后先等1秒，让多个线程同时竞争资源
    	                Thread.sleep(1000);
    	            }catch (InterruptedException e){
    	                e.printStackTrace();
    	            }
    				return "恭喜您，购买成功！";
    			}else{
    				return "获取库存失败。";
    			}
    		} finally {
    			//释放锁
    			redisLock.release(code, String.valueOf(time));
    			System.out.println("释放锁的时间戳："+String.valueOf(time));
    		}
    	}
    	
    	/**
    	 * 获取库存对象
    	 * @param code
    	 * @return
    	 */
    	@Override
    	public GoodsStore getGoodsStore(String code){
    		Optional<GoodsStore> optional = goodsStoreRespository.findById(code);
    		return optional.get();
    	}
    }

创建并实现测试的控制器

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Controller;
    import org.springframework.ui.Model;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PostMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.ResponseBody;
    import org.springframework.web.servlet.ModelAndView;
    
    import com.distributedlock.redis.facade.GoodsStoreFacade;
    
    @Controller
    @RequestMapping("/")
    public class TestController {
    	
    	@Autowired
    	private GoodsStoreFacade goodsStoreService;
    	
    	/**
    	 * 进入测试页面
    	 * @param model
    	 * @return
    	 */
    	@GetMapping("test")
    	public ModelAndView stepOne(Model model){
    		return new ModelAndView("test", "model", model);
    	}
    	
    	/**
    	 * 秒杀提交
    	 * @param code
    	 * @param num
    	 * @return
    	 */
    	@PostMapping("secKill")
    	@ResponseBody
    	public String secKill(@RequestParam(value="code",required=true) String code,@RequestParam(value="num",required=true) Integer num){
    		String reString = goodsStoreService.updateGoodsStore(code, num);
    		return reString;
    	}
    }

在templates目录创建test.html，实现点击秒杀功能。

<!DOCTYPE html>
    <html xmlns:th="http://www.thymeleaf.org"
    	xmlns:layout="http://www.ultraq.net.nz/thymeleaf/layout">
    <head>
    <meta charset="UTF-8">
    <title>Insert title here</title>
    <script th:src="@{/js/jquery-3.3.1.min.js}"></script>
    </head>
    <body>
    	<button id="btn_secKill">秒杀商品</button>
    	<div id="count_num"></div>
    	<div id="result"></div>
    	<script type="text/javascript">
    		var countNum = 0;
    		$(function(){
    			$("#btn_secKill").click(function(){
    				var json={"code":"2019053016502800101","num":2000};
    				for(var i = 0 ; i < 400 ; i++){
    					$.post("secKill",json,function(data){
    						if(data != "排队人数太多，请稍后再试."){
    							$("#result").append("<br />" + data + "<br />");
    						}else{
    							$("#result").append(data + " ");
    						}
    						if(data.indexOf("恭喜您，购买成功") != -1){
    							countNum += 2000;
    						}
    						$("#count_num").text("总共卖出："+countNum);
    					});
    				}
    			});
    		});
    	</script>
    </body>
    </html>

创建数据库并建表如下：  
![在这里插入图片描述][20190531185102101.png]  
**4、加锁情况测试**  
启动应用，访问http://localhost:8090/test  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5aGx3enk_size_16_color_FFFFFF_t_70 2]  
我们在程序中点击一次秒杀是模拟400个用户请求，但是即使这么多请求过去，在一秒钟的时间内因为我们一个线程完成购买够我们Sleep了1秒，所以1秒内也只有一个请求成功购买到商品，其他请求都因线程同步返回。

**5、不加锁情况测试**

修改GoodsStoreService如下：

import java.util.Optional;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import com.distributedlock.redis.entity.GoodsStore;
    import com.distributedlock.redis.facade.GoodsStoreFacade;
    import com.distributedlock.redis.lock.RedisLock;
    import com.distributedlock.redis.respository.GoodsStoreRespository;
    
    /**
     * 库存管理服务
     * @author user
     *
     */
    @Service
    public class GoodsStoreService implements GoodsStoreFacade{
    	@Autowired
    	private GoodsStoreRespository goodsStoreRespository;
    
    	/**
    	 * 根据产品编号更新库存
    	 * @param code
    	 * @return
    	 */
    	public String updateGoodsStore(String code,int count) {
    
    		}
    		try {
    			GoodsStore goodsStore = getGoodsStore(code);
    			if(goodsStore != null){
    				if(goodsStore.getStore() <= 0){
    					return "对不起，卖完了，库存为："+goodsStore.getStore();
    				}
    				if(goodsStore.getStore() < count){
    					return "对不起，库存不足，库存为："+goodsStore.getStore()+" 您的购买数量为："+count;
    				}
    				System.out.println("剩余库存："+goodsStore.getStore());
    				System.out.println("扣除库存："+count);
    				goodsStoreRespository.updateStore(code, count);
    				try{
    					//为了更好的测试多线程同时进行库存扣减，在进行数据更新之后先等1秒，让多个线程同时竞争资源
    	                Thread.sleep(1000);
    	            }catch (InterruptedException e){
    	                e.printStackTrace();
    	            }
    				return "恭喜您，购买成功！";
    			}else{
    				return "获取库存失败。";
    			}
    		} finally {
    			
    		}
    	}
    	
    	/**
    	 * 获取库存对象
    	 * @param code
    	 * @return
    	 */
    	@Override
    	public GoodsStore getGoodsStore(String code){
    		Optional<GoodsStore> optional = goodsStoreRespository.findById(code);
    		return optional.get();
    	}
    }

启动应用，访问http://localhost:8090/test  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5aGx3enk_size_16_color_FFFFFF_t_70 3]  
点击秒杀，在不加锁的情况下，库存多扣了，多秒了2000个商品，数据库的库存变成了-2000。  
![在这里插入图片描述][2019053118520375.png]

**6、小结**

本文我们简单介绍了为什么要使用分布式锁以及使用Spring Boot如何整合Redis实现分布式锁，希望对大家理解分布式锁有用。

Redis实现分布式锁，主要是基于setnx机制，Redis保证在同一时间，多个请求写入相同key的数据，只有一个请求可以成功写入。基于此，可以实现分布式锁。但是采用Redis实现，存在如下问题：

A. 2.0版本之前的Redis，setnx和expire两个命令是非原子性的操作，即极端情况下，在setnx命令执行完毕，expire命令执行之前，获得锁的服务突然宕机，那么就会造成其他节点无法获得锁而造成死锁。

B. 2.0版本之后，Redis通过set(key,value,expire)命令，将setnx和expire两个命令合并为一个原子操作，解决了A的问题，但是采用Redis实现分布式锁，加锁和释放锁都是在客户端完成，在特殊情况下，如果在设定的超时时间之内，拿到锁服务A未处理完毕，Redis在超时后自动释放锁，其他进程B就可以拿到锁，在进程B拿到锁后，进程A执行完毕，调用del会造成误删锁。所以在执行del的时候，还需要判断是不是自己的锁。

C. 在B的场景下，解决误删锁的情况，还是存在问题，那就是A服务没有执行完毕，B服务就获得了锁，这不符合分布式锁的逻辑，解决这个问题，那么客户端就需要开启一个守护线程，在超时之前如果服务还未处理完毕，要定时给锁续命。

D. Redis在分布式CAP模型中，集群模式下它是AP模型，只有单机情况下才是CP模型，但是分布式锁，需要保证的是CP模型，即一致性可分区容错，所以大型分布式场景下，Redis是不能作为分布式锁的实现方案的。

实例代码：https://download.csdn.net/download/zyhlwzy/11218477

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5aGx3enk_size_16_color_FFFFFF_t_70]: /images/20220124/d976c6d91e3b46b6be56bed9fc7d6e3a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5aGx3enk_size_16_color_FFFFFF_t_70 1]: /images/20220124/dbcfc1818d6645e797fbee60375c1b31.png
[20190531185102101.png]: /images/20220124/42bc4c8582c64c3eb867b475aa988ff0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5aGx3enk_size_16_color_FFFFFF_t_70 2]: /images/20220124/7f9b351516be4459bfdb73d19f23fbcb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5aGx3enk_size_16_color_FFFFFF_t_70 3]: /images/20220124/4bae516d2383458a99246fa093f113a1.png
[2019053118520375.png]: /images/20220124/7b713fd469a0471fafbfddf909e011c2.png