Ce que je veux partager avec vous aujourd'hui, c'est le verrouillage distribué Cet article utilise cinq cas, des diagrammes, une analyse de code source, etc.
Les verrous communs synchronisés, de verrouillage, etc. sont basés sur un seul JVM est implémenté, que devons-nous faire dans un scénario distribué ? A cette époque, des verrous distribués sont apparus. JVM的实现的，如果分布式场景下怎么办呢？这时候分布式锁就出现了。
关于分布式的实现方案，在业界流行的有三种：
1、基于数据库
2、基于Redis
3、基于Zookeeper
另外，还有使用etcd、consul
Redis🎜🎜3. Basé surZookeeper🎜🎜De plus, il y a aussi l'utilisation de etcd, consul. 🎜Le plus couramment utilisé en développement est Redis et Zookeeper propose deux solutions, et la plus complexe des deux solutions et la plus susceptible de causer des problèmes est Redis plan de mise en œuvre, nous allons donc aujourd'hui mettre Redis solutions d'implémentation. Redis和Zookeeper两种方案，并且两种方案中最复杂的，最容易出问题的就是Redis的实现方案，所以，我们今天就来把Redis实现方案都聊聊。
估计部分朋友还不太清楚分布式的使用场景，下面我简单罗列三种：
如下代码模拟了下单减库存的场景，我们分析下在高并发场景下会存在什么问题
@RestController
public class IndexController {
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 模拟下单减库存的场景
* @return
*/
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
return "end";
}
}假设在Redis
On estime que certains amis ne sont pas clairs sur les scénarios d'utilisation distribuée. Je vais brièvement énumérer trois types ci-dessous :
🎜
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));La valeur initiale du stock dans Redis est de 100. 🎜🎜Maintenant, il y a 5 clients qui demandent cette interface en même temps, et il peut y avoir une exécution simultanée🎜int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));这行代码，获取到的值都为100，紧跟着判断大于0后都进行-1操作，最后设置到redis 中的值都为99。但正常执行完成后redis中的值应为 95。
在遇到案例1的问题后，大部分人的第一反应都会想到加锁来控制事务的原子性，如下代码所示：
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
synchronized (this){
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
}
return "end";
}现在当有多个请求访问该接口时，同一时刻只有一个请求可进入方法体中进行库存的扣减，其余请求等候。
但我们都知道，synchronized 锁是属于JVM级别的，也就是我们俗称的“单机锁”。但现在基本大部分公司使用的都是集群部署，现在我们思考下以上代码在集群部署的情况下还能保证库存数据的一致性吗？
答案是不能，如上图所示，请求经Nginx分发后，可能存在多个服务同时从Redis中获取库存数据，此时只加synchronized （单机锁）是无效的，并发越高，出现问题的几率就越大。
setnx：将 key 的值设为 value，当且仅当 key 不存在。
若给定 key 已经存在，则 setnx 不做任何动作。
使用setnx实现简单的分布式锁：
/**
* 模拟下单减库存的场景
* @return
*/
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
// 使用 setnx 添加分布式锁
// 返回 true 代表之前redis中没有key为 lockKey 的值，并已进行成功设置
// 返回 false 代表之前redis中已经存在 lockKey 这个key了
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
if(!result){
// 代表已经加锁了
return "error_code";
}
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
// 释放锁
stringRedisTemplate.delete(lockKey);
return "end";
}我们知道 Redis 是单线程执行，现在再看案例2中的流程图时，哪怕高并发场景下多个请求都执行到了setnx的代码，redis会根据请求的先后顺序进行排列，只有排列在队头的请求才能设置成功。其它请求只能返回“error_code”。
当setnx设置成功后，可执行业务代码对库存扣减，执行完成后对锁进行释放。
我们再来思考下以上代码已经完美实现分布式锁了吗？能够支撑高并发场景吗？答案并不是，上面的代码还是存在很多问题的，离真正的分布式锁还差的很远。
我们分析一下，上面的代码存在的问题：
死锁：假如第一个请求在setnx加锁完成后，执行业务代码时出现了异常，那释放锁的代码就无法执行，后面所有的请求也都无法进行操作了。
针对死锁的问题，我们对代码再次进行优化，添加try-finally，在finally中添加释放锁代码，这样无论如何都会执行释放锁代码，如下所示：
/**
* 模拟下单减库存的场景
* @return
*/
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
try{
// 使用 setnx 添加分布式锁
// 返回 true 代表之前redis中没有key为 lockKey 的值，并已进行成功设置
// 返回 false 代表之前redis中已经存在 lockKey 这个key了
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
if(!result){
// 代表已经加锁了
return "error_code";
}
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
}finally {
// 释放锁
stringRedisTemplate.delete(lockKey);
}
return "end";
}经过改进后的代码是否还存在问题呢？我们思考正常执行的情况下应该是没有问题，但我们假设请求在执行到业务代码时服务突然宕机了，或者正巧你的运维同事重新发版，粗暴的 kill -9 掉了呢，那代码还能执行 finally 吗？
针对想到的问题，对代码再次进行优化，加入过期时间，这样即便出现了上述的问题，在时间到期后锁也会自动释放掉，不会出现“死锁”的情况。
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
try{
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
if(!result){
// 代表已经加锁了
return "error_code";
}
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
}finally {
// 释放锁
stringRedisTemplate.delete(lockKey);
}
return "end";
}现在我们再思考一下，给锁加入过期时间后就可以了吗？就可以完美运行不出问题了吗？
超时时间设置的10s真的合适吗？如果不合适设置多少秒合适呢？如下图所示
Supposons qu'il y ait trois demandes en même temps.
Nous pouvons maintenant voir le problème en simulant simplement 3 requêtes. Dans un scénario de concurrence vraiment élevée, le verrou peut être confronté à "toujours invalide" ou "permanent invalide".
Alors, où est le problème spécifique ? Résumé dans les points suivants :
Au vu du problème Nous réfléchissons à la solution correspondante :
Spring Boot集成RedissonÉtapes
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.6.5</version>
</dependency>@Bean
public RedissonClient redisson(){
// 单机模式
Config config = new Config();
config.useSingleServer().setAddress("redis://192.168.3.170:6379").setDatabase(0);
return Redisson.create(config);
}@RestController
public class IndexController {
@Autowired
private RedissonClient redisson;
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 模拟下单减库存的场景
* @return
*/
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
String lockKey = "product_001";
// 1.获取锁对象
RLock redissonLock = redisson.getLock(lockKey);
try{
// 2.加锁
redissonLock.lock(); // 等价于 setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
// 从redis 中拿当前库存的值
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
if(stock > 0){
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock",realStock + "");
System.out.println("扣减成功，剩余库存：" + realStock);
}else{
System.out.println("扣减失败，库存不足");
}
}finally {
// 3.释放锁
redissonLock.unlock();
}
return "end";
}
}
我们点击lock()方法，查看源码，最终看到以下代码
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
internalLockLeaseTime = unit.toMillis(leaseTime);
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);",
Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}没错，加锁最终执行的就是这段lua 脚本语言。
if (redis.call('exists', KEYS[1]) == 0) then
redis.call('hset', KEYS[1], ARGV[2], 1);
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;脚本的主要逻辑为：
这样来看其实和我们前面案例中的实现方法好像没什么区别，但实际上并不是。
这段lua脚本命令在Redis中执行时，会被当成一条命令来执行，能够保证原子性，故要不都成功，要不都失败。
我们在源码中看到Redssion的许多方法实现中很多都用到了lua脚本，这样能够极大的保证命令执行的原子性。
下面是Redisson锁自动“续命”源码：
private void scheduleExpirationRenewal(final long threadId) {
if (expirationRenewalMap.containsKey(getEntryName())) {
return;
}
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return 1; " +
"end; " +
"return 0;",
Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
future.addListener(new FutureListener<Boolean>() {
@Override
public void operationComplete(Future<Boolean> future) throws Exception {
expirationRenewalMap.remove(getEntryName());
if (!future.isSuccess()) {
log.error("Can't update lock " + getName() + " expiration", future.cause());
return;
}
if (future.getNow()) {
// reschedule itself
scheduleExpirationRenewal(threadId);
}
}
});
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {
task.cancel();
}
}这段代码是在加锁后开启一个守护线程进行监听。Redisson超时时间默认设置30s，线程每10s调用一次判断锁还是否存在，如果存在则延长锁的超时时间。
现在，我们再回过头来看看案例5中的加锁代码与原理图，其实完善到这种程度已经可以满足很多公司的使用了，并且很多公司也确实是这样用的。但我们再思考下是否还存在问题呢？例如以下场景：
Redis Dans le déploiement et l'utilisation réels, ils sont déployés dans des clusters. Dans les scénarios à haute concurrence, nous verrouillons. Après avoir écrit la clé sur le nœud maître, le maître plante avant d'être synchronisé avec le nœud esclave. Le original Le nœud esclave devient le nouveau nœud maître après l'élection. À ce moment-là, un problème d'échec de verrouillage peut survenir. Redis在实际部署使用时都是集群部署的，那在高并发场景下我们加锁，当把key写入到master节点后，master还未同步到slave节点时master宕机了，原有的slave节点经过选举变为了新的master节点，此时可能就会出现锁失效问题。Redis帮我们把请求进行了排队执行，也就是把我们的并行转为了串行。串行执行的代码肯定不存在并发问题了，但是程序的性能肯定也会因此受到影响。在思考解决方案时我们首先想到CAP原则（一致性、可用性、分区容错性），那么现在的Redis就是满足AP(可用性、分区容错性)，如果想要解决该问题我们就需要寻找满足CP(一致性、分区容错性)的分布式系统。首先想到的就是Zookeeper，Zookeeper🎜Grâce au mécanisme de mise en œuvre des verrous distribués, nous savons que dans les scénarios de concurrence élevée, seules les requêtes verrouillées avec succès peuvent continuer à traiter la logique métier. Ensuite, tout le monde vient verrouiller, mais un seul verrou réussit et les autres attendent. En fait, les verrous distribués et la concurrence élevée sont sémantiquement contradictoires. Bien que nos requêtes soient toutes concurrentes, Redis nous aide à mettre en file d'attente les demandes d'exécution, ce qui signifie convertir notre parallèle en série. Le code exécuté en série n'aura certainement pas de problèmes de concurrence, mais les performances du programme en seront certainement affectées.
Redis satisfait AP (disponibilité, tolérance de partition). Si nous voulons résoudre ce problème, nous devons trouver une solution qui satisfait CP (cohérence, tolérance aux pannes de partition) système distribué. La première chose qui me vient à l'esprit est Zookeeper , Le mécanisme de synchronisation des données inter-clusters de Zookeeper est que lorsque le nœud maître reçoit les données, il ne renvoie pas immédiatement de retour de réussite au client. Il synchronise d'abord les données avec les nœuds enfants, puis avertir le client une fois que plus de la moitié des nœuds ont terminé la synchronisation. Le terminal a reçu avec succès.Et si le nœud maître tombe en panne, selon Zookeeper的Zab协议（Zookeeper原子广播）重新选举的主节点一定是已经同步成功的。那么问题来了，Redisson与Zookeeper分布式锁我们如何选择呢？答案是如果并发量没有那么高，可以用Zookeeper来做分布式锁，但是它的并发能力远远不如Redis。如果你对并发要求比较高的话，那就用Redis，偶尔出现的主从架构锁失效的问题其实是可以容忍的。
关于第二个提升性能的问题，我们可以参考ConcurrentHashMap的锁分段技术的思想，例如我们代码的库存量当前为1000，那我们可以分为10段，每段100，然后对每段分别加锁，这样就可以同时执行10个请求的加锁与处理，当然有要求的同学还可以继续细分。但其实Redis的Qps已经达到10W+, cela est tout à fait suffisant dans des scénarios sans concurrence particulièrement élevée.
Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!
![[Web front-end] Démarrage rapide de Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
