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

1、为什么需要分布式锁?

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

而对于分布式系统来说,由于多个请求可能被分发到不同的机器上去处理,如果这多个请求都是对同一个资源进行操作,那么使用基本的Java多线程线程同步技术可能就解决不了这个问题。
在这里插入图片描述
如上图,请求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:页面模板;

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

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/> <!-- lookup parent from repository -->
	</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>

创建数据库并建表如下:
在这里插入图片描述
4、加锁情况测试
启动应用,访问http://localhost:8090/test
在这里插入图片描述
我们在程序中点击一次秒杀是模拟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
在这里插入图片描述
点击秒杀,在不加锁的情况下,库存多扣了,多秒了2000个商品,数据库的库存变成了-2000。
在这里插入图片描述

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

Logo

权威|前沿|技术|干货|国内首个API全生命周期开发者社区

更多推荐