本文主要介绍“如何理解分布式系统中基于Redis的分布式锁”。在日常操作中,我相信很多人对于如何理解分布式系统中基于Redis的分布式锁有疑问。边肖查阅了各种资料,整理出简单易用的操作方法,希望能帮助大家解决“如何理解分布式系统中基于Redis的分布式锁”的疑惑。接下来,请和边肖一起学习!
偶尔,新项目会有不平衡的账户。前技术老板离职时给出的解释是:经过调查,没有找到原因,后来又忙得没时间解决,可能是框架的原因.
既然项目已经送到手里,这样的问题必须解决。在梳理了所有的账务处理逻辑后,最终找到了原因:并发数据库操作导致的热账。关于这个问题,我们来谈谈分布式系统中基于Redis的分布式锁。顺便说一下,我们也将分解问题的原因和解决方案。
00-1010系统并发不高,也有热账户,但没那么严重。问题的根源在于系统架构设计,人为制造并发。场景如下:商家批量导入一批数据,系统会进行预处理,增加或减少账户余额。
此时,另一个计划任务也将扫描和更新帐户。而且同一个账号上的操作分布在各个系统中,出现了热账号。
为了解决这个问题,我们可以考虑将会计系统从架构层面分离出来,集中在一个系统上进行处理。所有数据库事务和执行顺序都由会计系统作为一个整体来处理。技术上可以通过锁定机制锁定热账户。
本文详细阐述了热账户分布式锁的实现。
00-1010在Java多线程环境中,通常有几种类型的锁可以使用:
JVM内存模型级锁常用的有:同步、锁等。
数据库锁,如乐观锁、悲观锁等。
分布式锁;
JVM内存级锁可以保证单一服务下的线程安全,比如多个线程访问/修改一个全局变量。然而,当系统部署在集群中时,JVM级别的本地锁是无能为力的。
原因分析
和上面的情况一样,热账户属于分布式系统中的共享资源,我们通常用数据库锁或者分布式锁来解决。
数据库锁分为乐观锁锁和悲观锁.锁
悲观锁是基于数据库(Mysql的InnoDB)提供的独占锁。在事务操作中,MySQL将通过select向查询结果集中的每一行数据添加排他锁.对于update语句,其他线程将阻止此记录的更新和删除。从而实现共享资源的顺序执行(修改);
乐观锁相对于悲观锁定。乐观锁定假设数据的一般情况不会引起冲突,因此在数据提交更新时,会正式检查数据是否存在冲突。如果有冲突,会把异常信息返回给用户,让用户自己决定怎么做。乐观锁定适用于读多写少的场景,可以提高程序的吞吐量。实现乐观锁定时,通常是基于记录状态或添加版本版本来实现的。
悲观锁曾用于10-1010项目,但失败了。这也是使用悲观锁时常见的误区。下面我们来分析一下。
正常使用悲观锁的过程:
通过选择锁定记录;用于更新;
计算新余额,修改金额并存储;
执行以释放锁;
常见错误的处理过程:
查询账户余额,计算新余额;
通过选择锁定记录;用于更新;
修改金额并存储;
执行以释放锁;
在错误的流程中,比如A和B服务查询的余额都是100,A扣50,B扣40,然后A锁定记录,更新数据库到50;A解除锁定后,B锁定该记录,并将数据库更新为60。显然,后者涵盖了前者的更新。解决方法是在计算新余额之前,扩大锁的范围并提前锁。
悲观锁通常会给数据库带来巨大的压力。在实践中,乐观锁或分布式锁通常根据场景使用。
让我们言归正传,谈谈基于Redis的分布式锁的实现。
锁的分析
ong>Redis分布式锁实战演习
这里以Spring Boot、Redis、Lua脚本为例来演示分布式锁的实现。为了简化处理,示例中Redis既承担了分布式锁的功能,也承担了数据库的功能。
场景构建
集群环境下,对同一个账户的金额进行操作,基本步骤:
-
从数据库读取用户金额;
-
程序修改金额;
-
再将最新金额存储到数据库;
下面从最初不加锁,不同步处理,逐步推演出最终的分布式锁。
基础集成及类构建
准备一个不加锁处理的基础业务环境。
首先在Spring Boot项目中引入相关依赖:
<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>
账户对应实体类UserAccount:
public class UserAccount { //用户ID private String userId; //账户内金额 private int amount; //添加账户金额 public void addAmount(int amount) { this.amount = this.amount + amount; } // 省略构造方法和getter/setter }
创建一个线程实现类AccountOperationThread:
public class AccountOperationThread implements Runnable { private final static Logger logger = LoggerFactory.getLogger(AccountOperationThread.class); private static final Long RELEASE_SUCCESS = 1L; private String userId; private RedisTemplate<Object, Object> redisTemplate; public AccountOperationThread(String userId, RedisTemplate<Object, Object> redisTemplate) { this.userId = userId; this.redisTemplate = redisTemplate; } @Override public void run() { noLock(); } /** * 不加锁 */ private void noLock() { try { Random random = new Random(); // 模拟线程进行业务处理 TimeUnit.MILLISECONDS.sleep(random.nextInt(100) + 1); } catch (InterruptedException e) { e.printStackTrace(); } //模拟数据库中获取用户账号 UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId); // 金额+1 userAccount.addAmount(1); logger.info(Thread.currentThread().getName() + " : user id : " + userId + " amount : " + userAccount.getAmount()); //模拟存回数据库 redisTemplate.opsForValue().set(userId, userAccount); } }
其中RedisTemplate的实例化交给了Spring Boot:
@Configuration public class RedisConfig { @Bean public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) { RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>(); redisTemplate.setConnectionFactory(redisConnectionFactory); Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class); ObjectMapper objectMapper = new ObjectMapper(); objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY); objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL); jackson2JsonRedisSerializer.setObjectMapper(objectMapper); // 设置value的序列化规则和 key的序列化规则 redisTemplate.setValueSerializer(jackson2JsonRedisSerializer); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.afterPropertiesSet(); return redisTemplate; } }
最后,再准备一个TestController来进行触发多线程的运行:
@RestController public class TestController { private final static Logger logger = LoggerFactory.getLogger(TestController.class); private static ExecutorService executorService = Executors.newFixedThreadPool(10); @Autowired private RedisTemplate<Object, Object> redisTemplate; @GetMapping("/test") public String test() throws InterruptedException { // 初始化用户user_001到Redis,账户金额为0 redisTemplate.opsForValue().set("user_001", new UserAccount("user_001", 0)); // 开启10个线程进行同步测试,每个线程为账户增加1元 for (int i = 0; i < 10; i++) { logger.info("创建线程i=" + i); executorService.execute(new AccountOperationThread("user_001", redisTemplate)); } // 主线程休眠1秒等待线程跑完 TimeUnit.MILLISECONDS.sleep(1000); // 查询Redis中的user_001账户 UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get("user_001"); logger.info("user id : " + userAccount.getUserId() + " amount : " + userAccount.getAmount()); return "success"; } }
执行上述程序,正常来说10个线程,每个线程加1,结果应该是10。但多执行几次,会发现,结果变化很大,基本上都要比10小。
[pool-1-thread-5] c.s.redis.thread.AccountOperationThread : pool-1-thread-5 : user id : user_001 amount : 1 [pool-1-thread-4] c.s.redis.thread.AccountOperationThread : pool-1-thread-4 : user id : user_001 amount : 1 [pool-1-thread-3] c.s.redis.thread.AccountOperationThread : pool-1-thread-3 : user id : user_001 amount : 1 [pool-1-thread-1] c.s.redis.thread.AccountOperationThread : pool-1-thread-1 : user id : user_001 amount : 1 [pool-1-thread-1] c.s.redis.thread.AccountOperationThread : pool-1-thread-1 : user id : user_001 amount : 2 [pool-1-thread-2] c.s.redis.thread.AccountOperationThread : pool-1-thread-2 : user id : user_001 amount : 2 [pool-1-thread-5] c.s.redis.thread.AccountOperationThread : pool-1-thread-5 : user id : user_001 amount : 2 [pool-1-thread-4] c.s.redis.thread.AccountOperationThread : pool-1-thread-4 : user id : user_001 amount : 3 [pool-1-thread-1] c.s.redis.thread.AccountOperationThread : pool-1-thread-1 : user id : user_001 amount : 4 [pool-1-thread-3] c.s.redis.thread.AccountOperationThread : pool-1-thread-3 : user id : user_001 amount : 5 [nio-8080-exec-1] c.s.redis.controller.TestController : user id : user_001 amount : 5
以上述日志为例,前四个线程都将值改为1,也就是后面三个线程都将前面的修改进行了覆盖,导致最终结果不是10,只有5。这显然是有问题的。
Redis同步锁实现
针对上面的情况,在同一个JVM当中,我们可以通过线程加锁来完成。但在分布式环境下,JVM级别的锁是没办法实现的,这里可以采用Redis同步锁实现。
基本思路:第一个线程进入时,在Redis中进记录,当后续线程过来请求时,判断Redis是否存在该记录,如果存在则说明处于锁定状态,进行等待或返回。如果不存在,则进行后续业务处理。
/** * 1.抢占资源时判断是否被锁。 * 2.如未锁则抢占成功且加锁,否则等待锁释放。 * 3.业务完成后释放锁,让给其它线程。 * <p> * 该方案并未解决同步问题,原因:线程获得锁和加锁的过程,并非原子性操作,可能会导致线程A获得锁,还未加锁时,线程B也获得了锁。 */ private void redisLock() { Random random = new Random(); try { TimeUnit.MILLISECONDS.sleep(random.nextInt(1000) + 1); } catch (InterruptedException e) { e.printStackTrace(); } while (true) { Object lock = redisTemplate.opsForValue().get(userId + ":syn"); if (lock == null) { // 获得锁 -> 加锁 -> 跳出循环 logger.info(Thread.currentThread().getName() + ":获得锁"); redisTemplate.opsForValue().set(userId + ":syn", "lock"); break; } try { // 等待500毫秒重试获得锁 TimeUnit.MILLISECONDS.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); } } try { //模拟数据库中获取用户账号 UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId); if (userAccount != null) { //设置金额 userAccount.addAmount(1); logger.info(Thread.currentThread().getName() + " : user id : " + userId + " amount : " + userAccount.getAmount()); //模拟存回数据库 redisTemplate.opsForValue().set(userId, userAccount); } } finally { //释放锁 redisTemplate.delete(userId + ":syn"); logger.info(Thread.currentThread().getName() + ":释放锁"); } }
在while代码块中,先判断对应用户ID是否在Redis中存在,如果不存在,则进行set加锁,如果存在,则跳出循环继续等待。
上述代码,看起来实现了加锁的功能,但当执行程序时,会发现与未加锁一样,依旧存在并发问题。原因是:获取锁和加锁的操作并不是原子的。比如两个线程发现lock都是null,都进行了加锁,此时并发问题依旧存在。
Redis原子性同步锁
针对上述问题,可将获取锁和加锁的过程原子化处理。基于spring-boot-data-redis提供的原子化API可以实现:
// 该方法使用了redis的指令:SETNX key value // 1.key不存在,设置成功返回value,setIfAbsent返回true; // 2.key存在,则设置失败返回null,setIfAbsent返回false; // 3.原子性操作; Boolean setIfAbsent(K var1, V var2);
上述方法的原子化操作是对Redis的setnx命令的封装,在Redis中setnx的使用如下实例:
redis> SETNX mykey "Hello" (integer) 1 redis> SETNX mykey "World" (integer) 0 redis> GET mykey "Hello"
第一次,设置mykey时,并不存在,则返回1,表示设置成功;第二次设置mykey时,已经存在,则返回0,表示设置失败。再次查询mykey对应的值,会发现依旧是第一次设置的值。也就是说redis的setnx保证了唯一的key只能被一个服务设置成功。
理解了上述API及底层原理,来看看线程中的实现方法代码如下:
/** * 1.原子操作加锁 * 2.竞争线程循环重试获得锁 * 3.业务完成释放锁 */ private void atomicityRedisLock() { //Spring data redis 支持的原子性操作 while (!redisTemplate.opsForValue().setIfAbsent(userId + ":syn", "lock")) { try { // 等待100毫秒重试获得锁 TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } } logger.info(Thread.currentThread().getName() + ":获得锁"); try { //模拟数据库中获取用户账号 UserAccount userAccount = (UserAccount) redisTemplate.opsForValue().get(userId); if (userAccount != null) { //设置金额 userAccount.addAmount(1); logger.info(Thread.currentThread().getName() + " : user id : " + userId + " amount : " + userAccount.getAmount()); //模拟存回数据库 redisTemplate.opsForValue().set(userId, userAccount); } } finally { //释放锁 redisTemplate.delete(userId + ":syn"); logger.info(Thread.currentThread().getName() + ":释放锁"); } }
再次执行代码,会发现结果正确了,也就是说可以成功的对分布式线程进行了加锁。
Redis分布式锁的死锁
虽然上述代码执行结果没问题,但如果应用异常宕机,没来得及执行finally中释放锁的方法,那么其他线程则永远无法获得这个锁。
此时可采用setIfAbsent的重载方法:
Boolean setIfAbsent(K var1, V var2, long var3, TimeUnit var5);
基于该方法,可以设置锁的过期时间。这样即便获得锁的线程宕机,在Redis中数据过期之后,其他线程可正常获得该锁。
示例代码如下:
private void atomicityAndExRedisLock() { try { //Spring data redis 支持的原子性操作,并设置5秒过期时间 while (!redisTemplate.opsForValue().setIfAbsent(userId + ":syn", System.currentTimeMillis() + 5000, 5000, TimeUnit.MILLISECONDS)) { // 等待100毫秒重试获得锁 logger.info(Thread.currentThread().getName() + ":尝试循环获取锁"); TimeUnit.MILLISECONDS.sleep(1000); } logger.info(Thread.currentThread().getName() + ":获得锁--------"); // 应用在这里宕机,进程退出,无法执行 finally; Thread.currentThread().interrupt(); // 业务逻辑... } catch (InterruptedException e) { e.printStackTrace(); } finally { //释放锁 if (!Thread.currentThread().isInterrupted()) { redisTemplate.delete(userId + ":syn"); logger.info(Thread.currentThread().getName() + ":释放锁"); } } }
业务超时及守护线程
上面添加了Redis所的超时时间,看似解决了问题,但又引入了新的问题。
比如,正常情况下线程A在5秒内可正常处理完业务,但偶发会出现超过5秒的情况。如果将超时时间设置为5秒,线程A获得了锁,但业务逻辑处理需要6秒。此时,线程A还在正常业务逻辑,线程B已经获得了锁。当线程A处理完时,有可能将线程B的锁给释放掉。
在上述场景中有两个问题点:
-
第一,线程A和线程B可能会同时在执行,存在并发问题。
-
第二,线程A可能会把线程B的锁给释放掉,导致一系列的恶性循环。
当然,可以通过在Redis中设置value值来判断锁是属于线程A还是线程B。但仔细分析会发现,这个问题的本质是因为线程A执行业务逻辑耗时超出了锁超时的时间。
那么就有两个解决方案了:
-
第一,将超时时间设置的足够长,确保业务代码能够在锁释放之前执行完成;
-
第二,为锁添加守护线程,为将要过期释放但未释放的锁增加时间;
第一种方式需要全行大多数情况下业务逻辑的耗时,进行超时时间的设定。
第二种方式,可通过如下守护线程的方式来动态增加锁超时时间。
public class DaemonThread implements Runnable { private final static Logger logger = LoggerFactory.getLogger(DaemonThread.class); // 是否需要守护 主线程关闭则结束守护线程 private volatile boolean daemon = true; // 守护锁 private String lockKey; private RedisTemplate<Object, Object> redisTemplate; public DaemonThread(String lockKey, RedisTemplate<Object, Object> redisTemplate) { this.lockKey = lockKey; this.redisTemplate = redisTemplate; } @Override public void run() { try { while (daemon) { long time = redisTemplate.getExpire(lockKey, TimeUnit.MILLISECONDS); // 剩余有效期小于1秒则续命 if (time < 1000) { logger.info("守护进程: " + Thread.currentThread().getName() + " 延长锁时间 5000 毫秒"); redisTemplate.expire(lockKey, 5000, TimeUnit.MILLISECONDS); } TimeUnit.MILLISECONDS.sleep(300); } logger.info(" 守护进程: " + Thread.currentThread().getName() + "关闭 "); } catch (InterruptedException e) { e.printStackTrace(); } } // 主线程主动调用结束 public void stop() { daemon = false; } }
上述线程每隔300毫秒获取一下Redis中锁的超时时间,如果小于1秒,则延长5秒。当主线程调用关闭时,守护线程也随之关闭。
主线程中相关代码实现:
private void deamonRedisLock() { //守护线程 DaemonThread daemonThread = null; //Spring data redis 支持的原子性操作,并设置5秒过期时间 String uuid = UUID.randomUUID().toString(); String value = Thread.currentThread().getId() + ":" + uuid; try { while (!redisTemplate.opsForValue().setIfAbsent(userId + ":syn", value, 5000, TimeUnit.MILLISECONDS)) { // 等待100毫秒重试获得锁 logger.info(Thread.currentThread().getName() + ":尝试循环获取锁"); TimeUnit.MILLISECONDS.sleep(1000); } logger.info(Thread.currentThread().getName() + ":获得锁----"); // 开启守护线程 daemonThread = new DaemonThread(userId + ":syn", redisTemplate); Thread thread = new Thread(daemonThread); thread.start(); // 业务逻辑执行10秒... TimeUnit.MILLISECONDS.sleep(10000); } catch (InterruptedException e) { e.printStackTrace(); } finally { //释放锁 这里也需要原子操作,今后通过 Redis + Lua 讲 String result = (String) redisTemplate.opsForValue().get(userId + ":syn"); if (value.equals(result)) { redisTemplate.delete(userId + ":syn"); logger.info(Thread.currentThread().getName() + ":释放锁-----"); } //关闭守护线程 if (daemonThread != null) { daemonThread.stop(); } } }
其中在获得锁之后,开启守护线程,在finally中将守护线程关闭。
基于Lua脚本的实现
在上述逻辑中,我们是基于spring-boot-data-redis提供的原子化操作来保证锁判断和执行的原子化的。在非Spring Boot项目中,则可以基于Lua脚本来实现。
首先定义加锁和解锁的Lua脚本及对应的DefaultRedisScript
对象,在RedisConfig
配置类中添加如下实例化代码:
@Configuration public class RedisConfig { //lock script private static final String LOCK_SCRIPT = " if redis.call('setnx',KEYS[1],ARGV[1]) == 1 " + " then redis.call('expire',KEYS[1],ARGV[2]) " + " return 1 " + " else return 0 end "; private static final String UNLOCK_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call" + "('del', KEYS[1]) else return 0 end"; // ... 省略部分代码 @Bean public DefaultRedisScript<Boolean> lockRedisScript() { DefaultRedisScript<Boolean> defaultRedisScript = new DefaultRedisScript<>(); defaultRedisScript.setResultType(Boolean.class); defaultRedisScript.setScriptText(LOCK_SCRIPT); return defaultRedisScript; } @Bean public DefaultRedisScript<Long> unlockRedisScript() { DefaultRedisScript<Long> defaultRedisScript = new DefaultRedisScript<>(); defaultRedisScript.setResultType(Long.class); defaultRedisScript.setScriptText(UNLOCK_SCRIPT); return defaultRedisScript; } }
再通过在AccountOperationThread
类中新建构造方法,将上述两个对象传入类中(省略此部分演示)。然后,就可以基于RedisTemplate
来调用了,改造之后的代码实现如下:
private void deamonRedisLockWithLua() { //守护线程 DaemonThread daemonThread = null; //Spring data redis 支持的原子性操作,并设置5秒过期时间 String uuid = UUID.randomUUID().toString(); String value = Thread.currentThread().getId() + ":" + uuid; try { while (!redisTemplate.execute(lockRedisScript, Collections.singletonList(userId + ":syn"), value, 5)) { // 等待1000毫秒重试获得锁 logger.info(Thread.currentThread().getName() + ":尝试循环获取锁"); TimeUnit.MILLISECONDS.sleep(1000); } logger.info(Thread.currentThread().getName() + ":获得锁----"); // 开启守护线程 daemonThread = new DaemonThread(userId + ":syn", redisTemplate); Thread thread = new Thread(daemonThread); thread.start(); // 业务逻辑执行10秒... TimeUnit.MILLISECONDS.sleep(10000); } catch (InterruptedException e) { logger.error("异常", e); } finally { //使用Lua脚本:先判断是否是自己设置的锁,再执行删除 // key存在,当前值=期望值时,删除key;key存在,当前值!=期望值时,返回0; Long result = redisTemplate.execute(unlockRedisScript, Collections.singletonList(userId + ":syn"), value); logger.info("redis解锁:{}", RELEASE_SUCCESS.equals(result)); if (RELEASE_SUCCESS.equals(result)) { if (daemonThread != null) { //关闭守护线程 daemonThread.stop(); logger.info(Thread.currentThread().getName() + ":释放锁---"); } } } }
其中while循环中加锁和finally中的释放锁都是基于Lua脚本来实现了。
Redis锁的其他因素
除了上述实例,在使用Redis分布式锁时,还可以考虑以下情况及方案。
Redis锁的不可重入
当线程在持有锁的情况下再次请求加锁,如果一个锁支持一个线程多次加锁,那么这个锁就是可重入的。如果一个不可重入锁被再次加锁,由于该锁已经被持有,再次加锁会失败。Redis可通过对锁进行重入计数,加锁时加 1,解锁时减 1,当计数归 0时释放锁。
可重入锁虽然高效但会增加代码的复杂性,这里就不举例说明了。
等待锁释放
有的业务场景,发现被锁则直接返回。但有的场景下,客户端需要等待锁释放然后去抢锁。上述示例就属于后者。针对等待锁释放也有两种方案:
-
客户端轮训:当未获得锁时,等待一段时间再重新获取,直到成功。上述示例就是基于这种方式实现的。这种方式的缺点也很明显,比较耗费服务器资源,当并发量大时会影响服务器的效率。
-
使用Redis的订阅发布功能:当获取锁失败时,订阅锁释放消息,获取锁成功后释放时,发送释放消息。
集群中的主备切换和脑裂
在Redis包含主从同步的集群部署方式中,如果主节点挂掉,从节点提升为主节点。如果客户端A在主节点加锁成功,指令还未同步到从节点,此时主节点挂掉,从节点升为主节点,新的主节点中没有锁的数据。这种情况下,客户端B就可能加锁成功,从而出现并发的场景。
当集群发生脑裂时,Redis master节点跟slave 节点和 sentinel 集群处于不同的网络分区。sentinel集群无法感知到master的存在,会将 slave 节点提升为 master 节点,此时就会存在两个不同的 master 节点。从而也会导致并发问题的出现。Redis Cluster集群部署方式同理。
到此,关于“如何理解分布式系统下基于Redis的分布式锁”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注网站,小编会继续努力为大家带来更多实用的文章!
内容来源网络,如有侵权,联系删除,本文地址:https://www.230890.com/zhan/55807.html