- 1《python每天一小段》--（9）使用Paramiko库，批量操作linux服务器_怎么在jupyter notebook中安装paramiko
- 2神经网络与深度学习课程作业-4-循环神经网络与NLP_关于循环神经网络习题
- 3如何将D3.js的强交互延续到Python中？
- 4第六章树和二叉树作业1—二叉树_已知一棵二叉树的树形如下图所示,若其后序遍历序列为 f, d, b, e, c, a,则其先(前
- 5自然语言处理之文本情感分析_自然语言处理和情感分析
- 6【看表情包学Linux】(9) 缓冲区的概念 | Git 三板斧 | 实现简易进度条_git缓冲区大小是什么
- 7机器人系统ros2-开发实践03-监听节点的参数变化（C++）
- 8解锁Apache Shiro：新手友好的安全框架指南(一)——整体架构与身份认证_apache shiro的配置包括安全管理器(3)
- 9自动化机器学习流水线：基于Spring Boot与AI机器学习技术的融合探索_springboot集成ai
- 10srilm使用
spring cloud 实现分布式锁--黑马程序员笔记_springcloud分布式锁
赞
踩
1. 概述
首先，要了解什么是分布式锁，首先要了解什么是线程锁和进程锁。
1.线程锁
主要用来给方法、代码块加锁。当某个方法或代码使用锁，在同一时刻仅有一个线程执行该方法或该代码段。线程锁只在同一JVM中有效果，因为线程锁的实现在根本上是依靠线程之间共享内存实现的，比如Synchronized、Lock等。
2.进程锁
为了控制同一操作系统中多个进程访问某个共享资源，因为进程具有独立性，各个进程无法访问其他进程的资源，因此无法通过synchronized等线程锁实现进程锁。
3.分布式锁
本地锁只能控制所在虚拟机中的线程同步执行，现在要实现分布式环境下所有虚拟机中的线程去同步执行就需要让多个虚拟机去共用一个锁，虚拟机可以分布式部署，锁也可以分布式部署。如下图：
虚拟机都去抢占同一个锁，锁是一个单独的程序提供加锁、解锁服务，谁抢到锁谁去查询数据库。
该锁已不属于某个虚拟机，而是分布式部署，由多个虚拟机所共享，这种锁叫分布式锁。
2. 为什么要用到分布式锁
如果将同步锁的程序分布式部署在多个虚拟机上则无法保证同一个key只会查询一次数据库,如下图：
一个同步锁程序只能保证同一个虚拟机中多个线程只有一个线程去数据库，如果高并发通过网关负载均衡转发给各个虚拟机，此时就会存在多个线程去查询数据库情况，因为虚拟机中的锁只能保证该虚拟机自己的线程去同步执行，无法跨虚拟机保证同步执行。
我们将虚拟机内部的锁叫本地锁，本地锁只能保证所在虚拟机的线程同步执行。
3. 分布式锁的实现方案
3.1 基于数据库实现分布锁
利用数据库主键唯一性的特点，或利用数据库唯一索引的特点，多个线程同时去插入相同的记录，谁插入成功谁就抢到锁。
3.2 使用zookeeper实现
zookeeper是一个分布式协调服务，主要解决分布式程序之间的同步的问题。
zookeeper的结构类似的文件目录，多线程向zookeeper创建一个子目录(节点)只会有一个创建成功，利用此特点可以实现分布式锁，谁创建该结点成功谁就获得锁。
3.3 基于redis实现锁
redis提供了分布式锁的实现方案，比如：SETNX、set nx、redisson等。
redis实现分布式锁的方案可以在redis.cn网站查阅，地址http://www.redis.cn/commands/set.html
3.3.1 SETNX
3.3.1.1 概述
SETNX命令的工作过程是去set一个不存在的key，多个线程去设置同一个key只会有一个线程设置成功，设置成功的的线程拿到锁。
使用命令
SET resource-name anystring NX EX max-lock-time 即可实现。
NX：表示key不存在才设置成功。
EX：设置过期时间
- 1
- 2
- 3
启动三个ssh客户端 , 连接redis: docker exec -it redis redis-cli (这里使用的是docker容器技术)
先认证: auth redis
同时向三个客户端发送测试命令如下：
表示设置lock001锁，value为001，过期时间为30秒
SET lock001 001 NX EX 30
- 1
命令发送成功，观察三个ssh客户端发现只有一个设置成功，其它两个设置失败，设置成功的请求表示抢到了lock001锁。
3.3.1.2 如何在代码中使用Set nx去实现分布锁呢？
使用spring-boot-starter-data-redis 提供的api即可实现set nx。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
<version>2.6.2</version>
</dependency>
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
添加依赖后，在bean中注入restTemplate。
我们先分析一段伪代码如下：
if(缓存中有){
返回缓存中的数据
}else{
获取分布式锁
if(获取锁成功）{
try{
查询数据库
}finally{
释放锁
}
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
获取分布式锁
使用redisTemplate.opsForValue().setIfAbsent(key,vaue)获取锁
- 1
这里考虑一个问题，当set nx一个key/value成功后，这个key(就是锁)需要设置过期时间吗？
如果不设置过期时间当获取到了锁却没有执行finally这个锁将会一直存在，其它线程无法获取这个锁。
所以执行set nx时要指定过期时间，即使用如下的命令
SET resource-name anystring NX EX max-lock-time
- 1
具体调用的方法是：
redisTemplate.opsForValue().setIfAbsent(K var1, V var2, long var3, TimeUnit var5)
- 1
3.3.1.3如何释放锁
释放锁分为两种情况：key到期自动释放，手动删除。
- key到期自动释放的方法
因为锁设置了过期时间，key到期会自动释放，但是会存在一个问题就是 查询数据库等操作还没有执行完时key到期了，此时其它线程就抢到锁了，最终重复查询数据库执行了重复的业务操作。
怎么解决这个问题？
可以将key的到期时间设置的长一些，足以执行完成查询数据库并设置缓存等相关操作。
如果这样效率会低一些，另外这个时间值也不好把控。
- 手动删除锁
如果是采用手动删除锁可能和key到期自动删除有所冲突，造成删除了别人的锁。
比如：当查询数据库等业务还没有执行完时key过期了，此时其它线程占用了锁，当上一个线程执行查询数据库等业务操作完成后手动删除锁就把其它线程的锁给删除了。
要解决这个问题可以采用删除锁之前判断是不是自己设置的锁，伪代码如下：
if(缓存中有){ 返回缓存中的数据 }else{ 获取分布式锁: set lock 01 NX if(获取锁成功）{ try{ 查询数据库 }finally{ if(redis.call("get","lock")=="01"){ 释放锁: redis.call("del","lock") } } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
以上代码第11行到13行非原子性，也会导致删除其它线程的锁。
查看文档上的说明：http://www.redis.cn/commands/set.html
在调用setnx命令设置key/value时，每个线程设置不一样的value值，这样当线程去删除锁时可以先根据key查询出来判断是不是自己当时设置的vlaue，如果是则删除。
这整个操作是原子的，实现方法就是去执行上边的lua脚本。
Lua 是一个小巧的脚本语言，redis在2.6版本就支持通过执行Lua脚本保证多个命令的原子性。
什么是原子性？
这些指令要么全成功要么全失败。
以上就是使用Redis Nx方式实现分布式锁，为了避免删除别的线程设置的锁需要使用redis去执行Lua脚本的方式去实现，这样就具有原子性，但是过期时间的值设置不存在不精确的问题。
3.3.2 Redisson实现分布式锁
3.3.2.1 什么是Redisson
再查阅 文档http://www.redis.cn/commands/set.html可以看到:
·
我们选用Java的实现方案 https://github.com/redisson/redisson
Redisson的文档地址：https://github.com/redisson/redisson/wiki/Table-of-Content
Redisson底层采用的是Netty 框架。支持Redis 2.8以上版本，支持Java1.6+以上版本。Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) 。
使用Redisson可以非常方便将Java本地内存中的常用数据结构的对象搬到分布式缓存redis中。
也可以将常用的并发编程工具如：AtomicLong、CountDownLatch、Semaphore等支持分布式。
使用RScheduledExecutorService 实现分布式调度服务。
支持数据分片，将数据分片存储到不同的redis实例中。
支持分布式锁，基于Java的Lock接口实现分布式锁，方便开发。
下边使用Redisson将Queue队列的数据存入Redis，实现一个排队及出队的接口。
3.3.2.2 使用redisson
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.11.2</version>
</dependency>
- 1
- 2
- 3
- 4
- 5
redis配置文件：
spring: redis: host: localhost # 主机地址 prot: 6378 # 端口 password: 123456 # 密码 database: 0 # 使用哪个库 lettuce: pool: max-active: 20 #连接池最大链接数默认值为8 max-wait: -1 #连接池最大阻塞时间（使用负值表示没有限制）默认为-1 max-idle: 10 #连接池中的最大空闲连接数 默认为8 min-idle: 0 #连接池中的最小空闲连接数 默认为8 timeout: 10000 # 超时时间 redisson: config: classpath:singleServerConfig.yaml #config: classpath:clusterServersConfig.yaml
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
singleServerConfig
--- singleServerConfig: #如果当前连接池里的连接数量超过了最小空闲连接数，而同时有连接空闲时间超过了该数值， #那么这些连接将会自动被关闭，并从连接池里去掉。时间单位是毫秒。 #默认值：10000 idleConnectionTimeout: 10000 pingTimeout: 1000 #同任何节点建立连接时的等待超时。时间单位是毫秒。 #默认值：10000 connectTimeout: 10000 #等待节点回复命令的时间。该时间从命令发送成功时开始计时。 #默认值：3000 timeout: 3000 #如果尝试达到 retryAttempts（命令失败重试次数） #仍然不能将命令发送至某个指定的节点时，将抛出错误。如果尝试在此限制之内发送成功， #则开始启用 timeout（命令等待超时） 计时。 #默认值：3 retryAttempts: 3 #在某个节点执行相同或不同命令时，连续失败failedAttempts（执行失败最大次数）时， #该节点将被从可用节点列表里清除，直到 reconnectionTimeout（重新连接时间间隔） 超时以后再次尝试。 #默认值：1500 retryInterval: 1500 #重新连接时间间隔 reconnectionTimeout: 3000 #执行失败最大次数 failedAttempts: 3 #密码 password: redis #数据库选择 select 4 database: 0 #每个连接的最大订阅数量。 #默认值：5 subscriptionsPerConnection: 5 #在Redis节点里显示的客户端名称。 clientName: null #在Redis节点 address: "redis://192.168.101.65:6379" #从节点发布和订阅连接的最小空闲连接数 #默认值：1 subscriptionConnectionMinimumIdleSize: 1 #用于发布和订阅连接的连接池最大容量。连接池的连接数量自动弹性伸缩。 #默认值：50 subscriptionConnectionPoolSize: 50 #节点最小空闲连接数 #默认值：32 connectionMinimumIdleSize: 32 #节点连接池大小 #默认值：64 connectionPoolSize: 64 #这个线程池数量被所有RTopic对象监听器，RRemoteService调用者和RExecutorService任务共同共享。 #默认值: 当前处理核数量 * 2 threads: 8 #这个线程池数量是在一个Redisson实例内，被其创建的所有分布式数据类型和服务， #以及底层客户端所一同共享的线程池里保存的线程数量。 #默认值: 当前处理核数量 * 2 nettyThreads: 8 #Redisson的对象编码类是用于将对象进行序列化和反序列化，以实现对该对象在Redis里的读取和存储。 #默认值: org.redisson.codec.JsonJacksonCodec codec: !<org.redisson.codec.JsonJacksonCodec> {} #传输模式 #默认值：TransportMode.NIO transportMode: "NIO"
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
redis集群配置clusterServersConfig.yaml
--- clusterServersConfig: #如果当前连接池里的连接数量超过了最小空闲连接数，而同时有连接空闲时间超过了该数值， #那么这些连接将会自动被关闭，并从连接池里去掉。时间单位是毫秒。 #默认值：10000 idleConnectionTimeout: 10000 #同任何节点建立连接时的等待超时。时间单位是毫秒。 #默认值：10000 connectTimeout: 10000 #等待节点回复命令的时间。该时间从命令发送成功时开始计时。 #默认值：3000 timeout: 3000 #如果尝试达到 retryAttempts（命令失败重试次数） #仍然不能将命令发送至某个指定的节点时，将抛出错误。如果尝试在此限制之内发送成功， #则开始启用 timeout（命令等待超时） 计时。 #默认值：3 retryAttempts: 3 #在某个节点执行相同或不同命令时，连续失败failedAttempts（执行失败最大次数）时， #该节点将被从可用节点列表里清除，直到 reconnectionTimeout（重新连接时间间隔） 超时以后再次尝试。 #默认值：1500 retryInterval: 1500 #密码 password: null #每个连接的最大订阅数量。 #默认值：5 subscriptionsPerConnection: 5 clientName: null #负载均衡算法类的选择 #默认值： org.redisson.connection.balancer.RoundRobinLoadBalancer #在使用多个Elasticache Redis服务节点的环境里，可以选用以下几种负载均衡方式选择一个节点： #org.redisson.connection.balancer.WeightedRoundRobinBalancer - 权重轮询调度算法 #org.redisson.connection.balancer.RoundRobinLoadBalancer - 轮询调度算法 #org.redisson.connection.balancer.RandomLoadBalancer - 随机调度算法 loadBalancer: !<org.redisson.connection.balancer.RoundRobinLoadBalancer> {} slaveSubscriptionConnectionMinimumIdleSize: 1 slaveSubscriptionConnectionPoolSize: 50 slaveConnectionMinimumIdleSize: 32 slaveConnectionPoolSize: 64 masterConnectionMinimumIdleSize: 32 masterConnectionPoolSize: 64 readMode: "SLAVE" nodeAddresses: - "redis://192.168.200:129:7001" - "redis://192.168.200.129:7002" - "redis://192.168.200.129:7003" scanInterval: 1000 threads: 0 nettyThreads: 0 codec: !<org.redisson.codec.JsonJacksonCodec> {} "transportMode":"NIO"
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
Redisson相比set nx实现分布式锁要简单的多，工作原理如下：
• 加锁机制
线程去获取锁，获取成功: 执行lua脚本，保存数据到redis数据库。
线程去获取锁，获取失败: 一直通过while循环尝试获取锁，获取成功后，执行lua脚本，保存数据到redis
• WatchDog自动延期看门狗机制
第一种情况：在一个分布式环境下，假如一个线程获得锁后，突然服务器宕机了，那么这个时候在一定时间后这个锁会自动释放，你也可以设置锁的有效时间(当不设置默认30秒时），这样的目的主要是防止死锁的发生
第二种情况：线程A业务还没有执行完，时间就过了，线程A 还想持有锁的话，就会启动一个watch dog后台线程，不断的延长锁key的生存时间。
• lua脚本-保证原子性操作
主要是如果你的业务逻辑复杂的话，通过封装在lua脚本中发送给redis，而且redis是单线程的，这样就保证这段复杂业务逻辑执行的原子性
具体使用RLock操作分布锁，RLock继承JDK的Lock接口，所以他有Lock接口的所有特性，比如lock、unlock、trylock等特性,同时它还有很多新特性：强制锁释放，带有效期的锁,。
public interface RLock { //----------------------Lock接口方法----------------------- /** * 加锁 锁的有效期默认30秒 */ void lock(); /** * 加锁 可以手动设置锁的有效时间 * * @param leaseTime 锁有效时间 * @param unit 时间单位 小时、分、秒、毫秒等 */ void lock(long leaseTime, TimeUnit unit); /** * tryLock()方法是有返回值的，用来尝试获取锁， * 如果获取成功，则返回true，如果获取失败（即锁已被其他线程获取），则返回false . */ boolean tryLock(); /** * tryLock(long time, TimeUnit unit)方法和tryLock()方法是类似的， * 只不过区别在于这个方法在拿不到锁时会等待一定的时间， * 在时间期限之内如果还拿不到锁，就返回false。如果如果一开始拿到锁或者在等待期间内拿到了锁，则返回true * * @param time 等待时间 * @param unit 时间单位 小时、分、秒、毫秒等 */ boolean tryLock(long time, TimeUnit unit) throws InterruptedException; /** * 比上面多一个参数，多添加一个锁的有效时间 * * @param waitTime 等待时间 * @param leaseTime 锁有效时间 * @param unit 时间单位 小时、分、秒、毫秒等 * waitTime 大于 leaseTime */ boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException; /** * 解锁 */ void unlock(); }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
lock()：
• 此方法为加锁，但是锁的有效期采用默认30秒
• 如果主线程未释放，且当前锁未调用unlock方法，则进入到watchDog机制
• 如果主线程未释放，且当前锁调用unlock方法，则直接释放锁
//注入RedissonClient @Autowired Redissonclient redissonclient; //Redisson分布式锁 public CoursePublish getCoursePublishCache(Long courseId){ //查询缓存 String jsonString = (String) redisTemplate.opsForValue().get("course:" + courseId); if(StringUtils.isNotEmpty(jsonString)){ if(jsonString.equals("null")){ return null; } CoursePublish coursePublish = JSON.parseObject(jsonString, CoursePublish.class); return coursePublish; }else{ //每门课程设置一个锁 RLock lock = redissonClient.getLock("coursequerylock:"+courseId);获取锁的key 拿课程为例，每门课程都有一个锁 //获取锁 lock.lock(); try { //再次查询缓存 jsonString = (String) redisTemplate.opsForValue().get("course:" + courseId); //缓存中有 if(StringUtils.isNotEmpty(jsonString)){ //直接返回数据 CoursePublish coursePublish = JSON.parseObject(jsonString, CoursePublish.class); return coursePublish; } System.out.println("=========从数据库查询=========="); try { //手动延迟，测试锁的续期功能 Thread.sleep( millis: 60000); }catch (InterruptedException e) { throw new RuntimeException(e); } //从数据库查询 CoursePublish coursePublish = getCoursePublish(courseId); redisTemplate.opsForValue().set("course:" + courseId, JSON.toJSONString(coursePublish),1,TimeUnit.DAYS); return coursePublish; }finally { //释放锁 lock.unlock(); } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
