一、redis实现分布式锁原理为什么需要分布式锁在聊分布式锁之前，有必要先解释一下，为什么需要分布式锁。
与分布式锁相对就的是单机锁，我们在写多线程程序时，避免同时操作一个共享变量产生数据问题，通常会使用一把锁来互斥以保证共享变量的正确性，其使用范围是在同一个进程中。如果换做是多个进程，需要同时操作一个共享资源，如何互斥呢？现在的业务应用通常是微服务架构，这也意味着一个应用会部署多个进程，多个进程如果需要修改mysql中的同一行记录，为了避免操作乱序导致脏数据，此时就需要引入分布式锁了。
想要实现分布式锁，必须借助一个外部系统，所有进程都去这个系统上申请加锁。这个外部系统必须具有互斥能力，也就是说，如果两个请求同时到达，系统只会成功地为一个进程加锁，而另一个进程会失败。这个外部系统可以是数据库，也可以是redis或zookeeper，但为了追求性能，我们通常会选择使用redis或zookeeper来做。
redis可以作为一个共享存储系统，多个客户端可以共享访问，因此可以被用来保存分布式锁。而且 redis 的读写性能高，可以应对高并发的锁操作场景。这篇文章的重点在于介绍如何使用redis实现分布式锁，并探讨在实现过程中可能会遇到的问题。
分布式锁如何实现作为分布式锁实现过程中的共享存储系统，redis可以使用键值对来保存锁变量，在接收和处理不同客户端发送的加锁和释放锁的操作请求。那么，键值对的键和值具体是怎么定的呢？我们要赋予锁变量一个变量名，把这个变量名作为键值对的键，而锁变量的值，则是键值对的值，这样一来，redis就能保存锁变量了，客户端也就可以通过redis的命令操作来实现锁操作。
想要实现分布式锁，必须要求redis有互斥的能力。可以使用setnx命令，其含义是set if not exist，即如果key不存在，才会设置它的值，否则什么也不做。实现一种分布式锁的方法是，两个客户端进程互斥地执行该命令。
以下展示了redis使用key/value对保存锁变量，以及两个客户端同时请求加锁的操作过程。
加锁操作完成后，加锁成功的客户端，就可以去操作共享资源，例如，修改mysql的某一行数据。操作完成后，还要及时释放锁，给后来者让出操作共享资源的机会。如何释放锁呢？直接使用del命令删除这个key即可。这个逻辑非常简单，整体的流程写成伪代码就是下面这样。
// 加锁setnx lock_key 1// 业务逻辑do things// 释放锁del lock_key
但是，以上实现存在一个很大的问题，当客户端1拿到锁后，如果发生下面的场景，就会造成死锁。
程序处理业务逻辑异常，没及时释放锁进程挂了，没机会释放锁
以上情况会导致已经获得锁的客户端一直占用锁，其他客户端永远无法获取到锁。
如何避免死锁为了解决以上死锁问题，最容易想到的方案是在申请锁时，在redis中实现时，给锁设置一个过期时间，假设操作共享资源的时间不会超过10s，那么加锁时，给这个key设置10s过期即可。
但以上操作还是有问题，加锁、设置过期时间是2条命令，有可能只执行了第一条，第二条却执行失败，例如：
1.setnx执行成功，执行expire时由于网络问题，执行失败
2.setnx执行成功，redis异常宕机，expire没有机会执行
3.setnx执行成功，客户端异常崩溃，expire没有机会执行
总之这两条命令如果不能保证是原子操作，就有潜在的风险导致过期时间设置失败，依旧有可能发生死锁问题。幸好在redis 2.6.12之后，redis扩展了set命令的参数，可以在set的同时指定expire时间，这条操作是原子的，例如以下命令是设置锁的过期时间为10秒。
set lock_key 1 ex 10 nx
至此，解决了死锁问题，但还是有其他问题。想像下面这个这样一种场景：
客户端1加锁成功，开始操作共享资源
客户端1操作共享资源耗时太久，超过了锁的过期时间，锁失效（锁被自动释放）
客户端2加锁成功，开始操作共享资源
客户端1操作共享资源完成，在finally块中手动释放锁，但此时它释放的是客户端2的锁。
这里存在两个严重的问题：
锁过期
释放了别人的锁
第1个问题是评估操作共享资源的时间不准确导致的，如果只是一味增大过期时间，只能缓解问题降低出现问题的概率，依旧无法彻底解决问题。原因在于客户端在拿到锁之后，在操作共享资源时，遇到的场景是很复杂的，既然是预估的时间，也只能是大致的计算，不可能覆盖所有导致耗时变长的场景。
第二个问题在于解锁操作是不够严谨的，因为它是一种不加区分地释放锁的操作，没有对锁的所有权进行检查。如何解决呢？
锁被别人给释放了解决办法是，客户端在加锁时，设置一个只有自己知道的唯一标识进去，例如可以是自己的线程id，如果是redis实现，就是set key unique_value ex 10 nx。之后在释放锁时，要先判断这把锁是否归自己持有，只有是自己的才能释放它。
//释放锁 比较unique_value是否相等，避免误释放if redis.get("key") == unique_value then return redis.del("key")
这里释放锁使用的是get + del两条命令，这时又会遇到原子性问题了。
客户端1执行get，判断锁是自己的
客户端2执行了set命令，强制获取到锁（虽然发生概念很低，但要严谨考虑锁的安全性）
客户端1执行del，却释放了客户端2的锁
由此可见，以上get + del两个命令还是必须原子的执行才行。怎样原子执行两条命令呢？答案是lua脚本，可以把以上逻辑写成lua脚本，让redis执行。因为redis处理每个请求是单线程执行的，在执行一个lua脚本时其它请求必须等待，直到这个lua脚本处理完成，这样一来get+del之间就不会有其他命令执行了。
以下是使用lua脚本（unlock.script）实现的释放锁操作的伪代码，其中，keys[1]表示lock_key，argv[1]是当前客户端的唯一标识，这两个值都是我们在执行 lua脚本时作为参数传入的。
//lua脚本语言，释放锁 比较unique_value是否相等，避免误释放if redis.call("get",keys[1]) == argv[1] then return redis.call("del",keys[1])else return 0end
最后我们执行以下命令，即可
redis-cli --eval unlock.script lock_key , unique_value
这样一路优先下来，整个加锁、解锁流程就更严谨了，先小结一下，基于redis实现的分布式锁，一个严谨的流程如下：
加锁时要设置过期时间set lock_key unique_value ex expire_time nx
操作共享资源
释放锁：lua脚本，先get判断锁是否归属自己，再del释放锁
有了这个严谨的锁模型，我们还需要重新思考之前的那个问题，锁的过期时间不好评估怎么办。
如何确定锁的过期时间前面提到过，过期时间如果评估得不好，这个锁就会有提前过期的风险，一种妥协的解决方案是，尽量冗余过期时间，降低锁提前过期的概率，但这个方案并不能完美解决问题。是否可以设置这样的方案，加锁时，先设置一个预估的过期时间，然后开启一个守护线程，定时去检测这个锁的失效时间，如果锁快要过期了，操作共享资源还未完成，那么就自动对锁进行续期，重新设置过期时间。
redisson是一个已封装好这些工作的库，可以说是一种非常优秀的解决方案。redisson是一个java语言实现的redis sdk客户端，在使用分布式锁时，它就采用了自动续期的方案来避免锁过期，这个守护线程我们一般叫它看门狗线程。这个sdk提供的api非常友好，它可以像操作本地锁一样操作分布式锁。客户端一旦加锁成功，就会启动一个watch dog看门狗线程，它是一个后台线程，会每隔一段时间（这段时间的长度与设置的锁的过期时间有关）检查一下，如果检查时客户端还持有锁key（也就是说还在操作共享资源），那么就会延长锁key的生存时间。
那如果客户端在加锁成功后就宕机了呢？宕机了那么看门狗任务就不存在了，也就无法为锁续期了，锁到期自动失效。
redis的部署方式对锁的影响上面讨论的情况，都是锁在单个redis 实例中可能产生的问题，并没有涉及到redis的部署架构细节。
redis发展到现在，几种常见的部署架构有：
单机模式；
主从模式；
哨兵（sentinel）模式；
集群模式；
我们使用redis时，一般会采用主从集群+哨兵的模式部署，哨兵的作用就是监测redis节点的运行状态。普通的主从模式，当master崩溃时，需要手动切换让slave成为master，使用主从+哨兵结合的好处在于，当master异常宕机时，哨兵可以实现故障自动切换，把slave提升为新的master，继续提供服务，以此保证可用性。那么当主从发生切换时，分布式锁依旧安全吗？
想像这样的场景：
客户端1在master上执行set命令，加锁成功
此时，master异常宕机，set命令还未同步到slave上（主从复制是异步的）
哨兵将slave提升为新的master，但这个锁在新的master上丢失了，导致客户端2来加锁成功了，两个客户端共同操作共享资源
可见，当引入redis副本后，分布式锁还是可能受到影响。即使redis通过sentinel保证高可用，如果这个master节点由于某些原因发生了主从切换，那么就会出现锁丢失的情况。
集群模式+redlock实现高可靠的分布式锁
为了避免redis实例故障而导致的锁无法工作的问题，redis的开发者 antirez提出了分布式锁算法redlock。redlock算法的基本思路，是让客户端和多个独立的redis实例依次请求加锁，如果客户端能够和半数以上的实例成功地完成加锁操作，那么我们就认为，客户端成功地获得分布式锁了，否则加锁失败。这样一来，即使有单个redis实例发生故障，因为锁变量在其它实例上也有保存，所以，客户端仍然可以正常地进行锁操作，锁变量并不会丢失。
来具体看下redlock算法的执行步骤。redlock算法的实现要求redis采用集群部署模式，无哨兵节点，需要有n个独立的redis实例（官方推荐至少5个实例）。接下来，我们可以分成3步来完成加锁操作。
第一步是，客户端获取当前时间。
第二步是，客户端按顺序依次向n个redis实例执行加锁操作。
这里的加锁操作和在单实例上执行的加锁操作一样，使用set命令，带上nx、ex/px选项，以及带上客户端的唯一标识。当然，如果某个redis实例发生故障了，为了保证在这种情况下，redlock算法能够继续运行，我们需要给加锁操作设置一个超时时间。如果客户端在和一个redis实例请求加锁时，一直到超时都没有成功，那么此时，客户端会和下一个redis实例继续请求加锁。一般需要将加锁操作的超时时间设置为锁的有效时间的一小部分，通常约为几十毫秒。
第三步是，一旦客户端完成了和所有redis实例的加锁操作，客户端就要计算整个加锁过程的总耗时。
客户端只有在满足两个条件时，才能认为是加锁成功，条件一是客户端从超过半数（大于等于 n/2+1）的redis实例上成功获取到了锁；条件二是客户端获取锁的总耗时没有超过锁的有效时间。
为何只有在大多数实例加锁成功时才能算操作成功？事实上，多个redis实例一起使用组成了一个分布式系统。在分布式系统中总会出现异常节点，所以在谈论分布式系统时，需要考虑异常节点达到多少个，也依旧不影响整个系统的正确运行。这是一个分布式系统的容错问题，这个问题的结论是：如果只存在故障节点，只要大多数节点正常，那么整个系统依旧可以提供正确服务。
在满足了这两个条件后，我们需要重新计算这把锁的有效时间，计算的结果是锁的最初有效时间减去客户端为获取锁的总耗时。如果锁的有效时间已经来不及完成共享数据的操作了，我们可以释放锁，以免出现还没完成共享资源操作，锁就过期了的情况。
当然，如果客户端在和所有实例执行完加锁操作后，没能同时满足这两个条件，那么，客户端就要向所有redis节点发起释放锁的操作。为什么释放锁，要操作所有的节点呢，不能只操作那些加锁成功的节点吗？因为在某一个redis节点加锁时，可能因为网络原因导致加锁失败，例如一个客户端在一个redis实例上加锁成功，但在读取响应结果时由于网络问题导致读取失败，那这把锁其实已经在redis上加锁成功了。所以释放锁时，不管之前有没有加锁成功，需要释放所有节点上的锁以保证清理节点上的残留的锁。
在redlock算法中，释放锁的操作和在单实例上释放锁的操作一样，只要执行释放锁的 lua脚本就可以了。如果n个redis实例中超过一半的实例正常工作，就能确保分布式锁正常运作。为了提高分布式锁的可靠性，您可以在实际业务应用中使用redlock算法。
二、代码实现redis分布式锁1.springboot整合redis用到最多的当然属于我们的老朋友redistemplate，pom依赖如下：<!-- springboot整合redis --><dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-data-redis</artifactid></dependency>
2.redis配置类:package com.example.redisdemo.config;import com.fasterxml.jackson.annotation.jsonautodetect;import com.fasterxml.jackson.annotation.propertyaccessor;import com.fasterxml.jackson.databind.deserializationfeature;import com.fasterxml.jackson.databind.objectmapper;import org.springframework.context.annotation.bean;import org.springframework.context.annotation.configuration;import org.springframework.data.redis.connection.lettuce.lettuceconnectionfactory;import org.springframework.data.redis.core.redistemplate;import org.springframework.data.redis.serializer.jackson2jsonredisserializer;import org.springframework.data.redis.serializer.redisserializer;import org.springframework.data.redis.serializer.stringredisserializer;/** * @description: redis配置类 * @author keson * @date 21:20 2022/11/14 * @param * @return * @version 1.0 */@configurationpublic class redisconfig { @bean public redistemplate<string, object> redistemplate(lettuceconnectionfactory lettuceconnectionfactory) { // 设置序列化 jackson2jsonredisserializer<object> jackson2jsonredisserializer = new jackson2jsonredisserializer<object>(object.class); objectmapper om = new objectmapper(); om.setvisibility(propertyaccessor.all, jsonautodetect.visibility.any); om.configure(deserializationfeature.fail_on_unknown_properties, false); jackson2jsonredisserializer.setobjectmapper(om); // 配置redistemplate redistemplate<string, object> redistemplate = new redistemplate<string, object>(); redistemplate.setconnectionfactory(lettuceconnectionfactory); redisserializer<?> stringserializer = new stringredisserializer(); redistemplate.setkeyserializer(stringserializer);// key序列化 redistemplate.setvalueserializer(jackson2jsonredisserializer);// value序列化 redistemplate.sethashkeyserializer(stringserializer);// hash key序列化 redistemplate.sethashvalueserializer(jackson2jsonredisserializer);// hash value序列化 redistemplate.afterpropertiesset(); return redistemplate; }}
3.service层面package com.example.redisdemo.service;import com.example.redisdemo.entity.customerbalance;import java.util.concurrent.callable;/** * @author keson * @version 1.0 * @description: todo * @date 2022/11/14 15:12 */public interface redisservice { <t> t callwithlock(customerbalance customerbalance, callable<t> callable) throws exception;}
package com.example.redisdemo.service.impl;import com.example.redisdemo.entity.customerbalance;import com.example.redisdemo.service.redisservice;import lombok.extern.slf4j.slf4j;import org.springframework.beans.factory.annotation.autowired;import org.springframework.data.redis.connection.redisstringcommands;import org.springframework.data.redis.connection.returntype;import org.springframework.data.redis.core.rediscallback;import org.springframework.data.redis.core.redistemplate;import org.springframework.data.redis.core.types.expiration;import org.springframework.stereotype.service;import java.nio.charset.standardcharsets;import java.util.uuid;import java.util.concurrent.callable;import java.util.concurrent.timeunit;/** * @author keson * @version 1.0 * @description: todo redis实现分布式锁 * @date 2022/11/14 15:13 */@service@slf4jpublic class redisserviceimpl implements redisservice { //设置默认过期时间 private final static int default_lock_expiry_time = 20; //自定义lock key前缀 private final static string lock_prefix = "lock:customer_balance"; @autowired private redistemplate redistemplate; @override public <t> t callwithlock(customerbalance customerbalance, callable<t> callable) throws exception{ //自定义lock key string lockkey = getlockkey(customerbalance.getcustomernumber(), customerbalance.getsubaccountnumber(), customerbalance.getcurrencycode()); //将uuid当做value，确保唯一性 string lockreference = uuid.randomuuid().tostring(); try { if (!lock(lockkey, lockreference, default_lock_expiry_time, timeunit.seconds)) { throw new exception("lock加锁失败"); } return callable.call(); } finally { unlock(lockkey, lockreference); } } //定义lock key string getlockkey(string customernumber, string subaccountnumber, string currencycode) { return string.format("%s:%s:%s:%s", lock_prefix, customernumber, subaccountnumber, currencycode); } //redis加锁 private boolean lock(string key, string value, long timeout, timeunit timeunit) { boolean locked; try { //set_if_absent --> nx: only set the key if it does not already exist. //set_if_present --> xx: only set the key if it already exist. locked = (boolean) redistemplate.execute((rediscallback<boolean>) connection -> connection.set(key.getbytes(standardcharsets.utf_8), value.getbytes(standardcharsets.utf_8), expiration.from(timeout, timeunit), redisstringcommands.setoption.set_if_absent)); } catch (exception e) { log.error("lock failed for redis key: {}, value: {}", key, value); locked = false; } return locked != null && locked; } //redis解锁 private boolean unlock(string key, string value) { try { //使用lua脚本保证删除的原子性，确保解锁 string script = "if redis.call('get', keys[1]) == argv[1] " + "then return redis.call('del', keys[1]) " + "else return 0 end"; boolean unlockstate = (boolean) redistemplate.execute((rediscallback<boolean>) connection -> connection.eval(script.getbytes(), returntype.boolean, 1, key.getbytes(standardcharsets.utf_8), value.getbytes(standardcharsets.utf_8))); return unlockstate == null || !unlockstate; } catch (exception e) { log.error("unlock failed for redis key: {}, value: {}", key, value); return false; } }}
4.业务调用实现分布式锁示例： @override public int updatebyid(customerbalance customerbalance) throws exception { return redisservice.callwithlock(customerbalance, ()-> customerbalancemapper.updatebyid(customerbalance)); }
以上就是怎么在springboot中使用redis实现分布式锁的详细内容。