Nous savons tous que les verrous distribués doivent être utilisés dans un environnement distribué. Alors, de quelles caractéristiques les verrous distribués ont-ils besoin ? Comment verrouiller Redis autonome ? Quels sont les pièges du verrouillage du cluster Redis ? Ne vous inquiétez pas, dévoilons étape par étape le voile du verrouillage distribué Redis.
Caractéristiques des verrous distribués
Un seul fil peut maintenir le verrou en toutes circonstances.
L'environnement de cluster Redis ne peut pas échouer à acquérir ou libérer des verrous car un nœud est en panne. [Recommandations associées : Tutoriel vidéo Redis]
doit avoir un mécanisme de contrôle du délai d'attente ou une opération d'annulation.
Verrouillez-le vous-même et relâchez-le vous-même. Les verrous ajoutés par d'autres ne peuvent pas être libérés.
Le même fil peut être verrouillé plusieurs fois.
Comment implémenter Redis sur une seule machine
Généralement, il est implémenté à l'aide du script setnx+lua.
Publiez le code directement
package com.fandf.test.redis;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.RandomUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
/**
* redis 单机锁
*
* @author fandongfeng
* @date 2023/3/29 06:52
*/
@Slf4j
@Service
public class RedisLock {
@Resource
RedisTemplate<String, Object> redisTemplate;
private static final String SELL_LOCK = "kill:";
/**
* 模拟秒杀
*
* @return 是否成功
*/
public String kill() {
String productId = "123";
String key = SELL_LOCK + productId;
//锁value,解锁时 用来判断当前锁是否是自己加的
String value = IdUtil.fastSimpleUUID();
//加锁 十秒钟过期 防死锁
Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, value, 10, TimeUnit.SECONDS);
if (!flag) {
return "加锁失败";
}
try {
String productKey = "good123";
//获取商品库存
Integer stock = (Integer) redisTemplate.opsForValue().get(productKey);
if (stock == null) {
//模拟录入数据， 实际应该加载时从数据库读取
redisTemplate.opsForValue().set(productKey, 100);
stock = 100;
}
if (stock <= 0) {
return "卖完了，下次早点来吧";
}
//扣减库存, 模拟随机卖出数量
int randomInt = RandomUtil.randomInt(1, 10);
redisTemplate.opsForValue().decrement(productKey, randomInt);
// 修改db,可以丢到队列里慢慢处理
return "成功卖出" + randomInt + "个，库存剩余" + redisTemplate.opsForValue().get(productKey) + "个";
} finally {
// //这种方法会存在删除别人加的锁的可能
// redisTemplate.delete(key);
// if(value.equals(redisTemplate.opsForValue().get(key))){
// //因为if条件的判断和 delete不是原子性的，
// //if条件判断成功后，恰好锁到期自己解锁
// //此时别的线程如果持有锁了，就会把别人的锁删除掉
// redisTemplate.delete(key);
// }
//使用lua脚本保证判断和删除的原子性
String luaScript =
"if (redis.call('get',KEYS[1]) == ARGV[1]) then " +
"return redis.call('del',KEYS[1]) " +
"else " +
"return 0 " +
"end";
redisTemplate.execute(new DefaultRedisScript<>(luaScript, Boolean.class), Collections.singletonList(key), value);
}
}
}Copier après la connexion
Effectuez des tests unitaires, simulez une centaine de threads pour effectuer un flash kill en même temps
package com.fandf.test.redis;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.parallel.Execution;
import org.springframework.boot.test.context.SpringBootTest;
import javax.annotation.Resource;
import static org.junit.jupiter.api.parallel.ExecutionMode.CONCURRENT;
/**
* @Description:
* @author: fandongfeng
* @date: 2023-3-24 16:45
*/
@SpringBootTest
class SignServiceTest {
@Resource
RedisLock redisLock;
@RepeatedTest(100)
@Execution(CONCURRENT)
public void redisLock() {
String result = redisLock.kill();
if("加锁失败".equals(result)) {
}else {
System.out.println(result);
}
}
}Copier après la connexion
Seuls trois threads ont saisi le verrou
成功卖出5个，库存剩余95个
成功卖出8个，库存剩余87个
成功卖出7个，库存剩余80个
Copier après la connexion
Qu'est-ce qui ne va pas avec le verrou redis ?
En général, il y en a deux :
- 1 Impossible de rentrer.
- 2. Afin d'éviter les blocages, nous ajouterons un délai d'expiration lors du verrouillage. Dans la plupart des cas, ce délai est basé sur l'évaluation de l'expérience des affaires existantes, mais en cas de blocage ou d'exception du programme, l'exécution prendra beaucoup de temps. . Après une longue période, le verrou sera automatiquement libéré à son expiration. À ce stade, si d'autres threads obtiennent le verrou et exécutent la logique, des problèmes peuvent survenir.
Alors, y a-t-il un moyen de résoudre ces deux problèmes ? Oui, parlons de Redisson
Redisson implémentant des verrous distribués
Qu'est-ce que Redisson ?
Redisson est une grille de données Java en mémoire (In-Memory Data Grid) implémentée sur la base de Redis. Il fournit non seulement une série d'objets Java communs distribués, mais fournit également de nombreux services distribués. Ceux-ci incluent (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 fournit le moyen le plus simple et le plus pratique d'utiliser Redis. Le but de Redisson est de promouvoir la séparation des préoccupations des utilisateurs (Separation of Concern) de Redis, afin que les utilisateurs puissent se concentrer davantage sur le traitement de la logique métier.
SpringBoot intègre Redesson
Integration est très simple, juste deux étapes
- pom introduit les dépendances
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
</dependency>Copier après la connexion
- application.yml ajoute de la configuration redis
spring:
application:
name: test
redis:
host: 127.0.0.1
port: 6379Copier après la connexion
it est également très simple à utiliser, il suffit d'injecter redissIlient
rreee
Les amis qui ne connaissent peut-être pas Redisson ne peuvent s’empêcher de poser des questions.
Quoi ? N'avez-vous pas besoin d'ajouter un délai d'expiration lors du verrouillage ? Cela entraînera-t-il une impasse ? Ne faut-il pas faire preuve de jugement pour déterminer si vous le détenez vous-même ?
Haha, ne vous inquiétez pas, nous dévoilerons le redisson étape par étape.
Suivi du code source Redisson lock()
Suivons la méthode lock() étape par étape pour examiner le code source (la version locale de Redisson est 3.20.0)
package com.fandf.test.redis;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* @author fandongfeng
*/
@Component
@Slf4j
public class RedissonTest {
@Resource
RedissonClient redissonClient;
public void test() {
RLock rLock = redissonClient.getLock("anyKey");
//rLock.lock(10, TimeUnit.SECONDS);
rLock.lock();
try {
// do something
} catch (Exception e) {
log.error("业务异常", e);
} finally {
rLock.unlock();
}
}
}Copier après la connexion
Affichez le verrou (-1, null, false ); méthode
//RedissonLock.class
@Override
public void lock() {
try {
lock(-1, null, false);
} catch (InterruptedException e) {
throw new IllegalStateException();
}
}Copier après la connexion
us Voyons comment il est verrouillé, qui est la méthode tryAcquire
private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
//获取当前线程id
long threadId = Thread.currentThread().getId();
//加锁代码块， 返回锁的失效时间
Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return;
}
CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
pubSub.timeout(future);
RedissonLockEntry entry;
if (interruptibly) {
entry = commandExecutor.getInterrupted(future);
} else {
entry = commandExecutor.get(future);
}
try {
while (true) {
ttl = tryAcquire(-1, leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
break;
}
// waiting for message
if (ttl >= 0) {
try {
entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
if (interruptibly) {
throw e;
}
entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
}
} else {
if (interruptibly) {
entry.getLatch().acquire();
} else {
entry.getLatch().acquireUninterruptibly();
}
}
}
} finally {
unsubscribe(entry, threadId);
}
// get(lockAsync(leaseTime, unit));
}Copier après la connexion
private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
//真假加锁方法 tryAcquireAsync
return get(tryAcquireAsync(waitTime, leaseTime, unit, threadId));
}Copier après la connexion
Le code ci-dessus contient la logique du verrouillage et du renouvellement du verrouillage. Jetons d'abord un coup d'œil au code de verrouillage
public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
super(commandExecutor, name);
this.commandExecutor = commandExecutor;
this.internalLockLeaseTime = commandExecutor.getServiceManager().getCfg().getLockWatchdogTimeout();
this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
}
private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
RFuture<Long> ttlRemainingFuture;
if (leaseTime > 0) {
ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
} else {
//waitTime和leaseTime都是-1，所以走这里
//过期时间internalLockLeaseTime初始化的时候赋值commandExecutor.getServiceManager().getCfg().getLockWatchdogTimeout();
//跟进去源码发现默认值是30秒， private long lockWatchdogTimeout = 30 * 1000;
ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
}
CompletionStage<Long> s = handleNoSync(threadId, ttlRemainingFuture);
ttlRemainingFuture = new CompletableFutureWrapper<>(s);
//加锁成功，开启子线程进行续约
CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {
// lock acquired
if (ttlRemaining == null) {
if (leaseTime > 0) {
//如果指定了过期时间，则不续约
internalLockLeaseTime = unit.toMillis(leaseTime);
} else {
//没指定过期时间，或者小于0，在这里实现锁自动续约
scheduleExpirationRenewal(threadId);
}
}
return ttlRemaining;
});
return new CompletableFutureWrapper<>(f);
}Copier après la connexion
Il. sera très clair ici. Redisson utilise le script Lua pour garantir l'atomicité des commandes.
redis.call('hexists', KEYS[1], ARGV[2]) Vérifiez si la valeur de la clé existe.
La commande Redis Hexists est utilisée pour vérifier si le champ spécifié de la table de hachage existe.
Si la table de hachage contient le champ donné, renvoyez 1. Si la table de hachage ne contient pas le champ donné ou si la clé n'existe pas, 0 est renvoyé.
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
"if ((redis.call('exists', KEYS[1]) == 0) " +
"or (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.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}Copier après la connexion
Lorsque la clé n'existe pas, ou contient déjà un champ donné (c'est-à-dire qu'elle a été verrouillée, c'est pour réaliser la réentrée), ajoutez directement 1 à la valeur du champ
La valeur de ce champ, qui est ARGV[2 ], obtenu par la méthode getLockName(threadId), regardons la valeur de ce champ
127.0.0.1:6379> hexists 123 uuid
(integer) 0
127.0.0.1:6379> hincrby 123 uuid 1
(integer) 1
127.0.0.1:6379> hincrby 123 uuid 1
(integer) 2
127.0.0.1:6379> hincrby 123 uuid 1
(integer) 3
127.0.0.1:6379> hexists 123 uuid
(integer) 1
127.0.0.1:6379> hgetall 123
1) "uuid"
2) "3"
127.0.0.1:6379>
Copier après la connexion
Vous l'aurez compris ici, le nom du champ est uuid + : + threadId
Regardons ensuite le renouvellement du verrou code planningExpirationRenewal( threadId);
protected String getLockName(long threadId) {
return id + ":" + threadId;
}
public RedissonBaseLock(CommandAsyncExecutor commandExecutor, String name) {
super(commandExecutor, name);
this.commandExecutor = commandExecutor;
this.id = commandExecutor.getServiceManager().getId();
this.internalLockLeaseTime = commandExecutor.getServiceManager().getCfg().getLockWatchdogTimeout();
this.entryName = id + ":" + name;
}
//commandExecutor.getServiceManager() 的id默认值
private final String id = UUID.randomUUID().toString();Copier après la connexion
Jetons un coup d'œil à la méthode renouvelerExpiration()
private void renewExpiration() {
ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ee == null) {
return;
}
//新建一个线程执行
Timeout task = commandExecutor.getServiceManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
if (ent == null) {
return;
}
Long threadId = ent.getFirstThreadId();
if (threadId == null) {
return;
}
//设置锁过期时间为30秒
CompletionStage<Boolean> future = renewExpirationAsync(threadId);
future.whenComplete((res, e) -> {
if (e != null) {
log.error("Can't update lock {} expiration", getRawName(), e);
EXPIRATION_RENEWAL_MAP.remove(getEntryName());
return;
}
//检查锁是还否存在
if (res) {
// reschedule itself 10后调用自己
renewExpiration();
} else {
//关闭续约
cancelExpirationRenewal(null);
}
});
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
//注意上行代码internalLockLeaseTime / 3，
//internalLockLeaseTime默认30s，那么也就是10s检查一次
ee.setTimeout(task);
}
//设置锁过期时间为internalLockLeaseTime 也就是30s lua脚本保证原子性
protected CompletionStage<Boolean> renewExpirationAsync(long threadId) {
return evalWriteAsync(getRawName(), 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.singletonList(getRawName()),
internalLockLeaseTime, getLockName(threadId));
}Copier après la connexion
OK，分析到这里我们已经知道了，lock()，方法会默认加30秒过期时间，并且开启一个新线程，每隔10秒检查一下，锁是否释放，如果没释放，就将锁过期时间设置为30秒，如果锁已经释放，那么就将这个新线程也关掉。
我们写个测试类看看
package com.fandf.test.redis;
import org.junit.jupiter.api.Test;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.boot.test.context.SpringBootTest;
import javax.annotation.Resource;
/**
* @Description:
* @author: fandongfeng
* @date: 2023-3-2416:45
*/
@SpringBootTest
class RedissonTest {
@Resource
private RedissonClient redisson;
@Test
public void watchDog() throws InterruptedException {
RLock lock = redisson.getLock("123");
lock.lock();
Thread.sleep(1000000);
}
}Copier après la connexion
查看锁的过期时间，及是否续约
127.0.0.1:6379> keys *
1) "123"
127.0.0.1:6379> ttl 123
(integer) 30
127.0.0.1:6379> ttl 123
(integer) 26
127.0.0.1:6379> ttl 123
(integer) 24
127.0.0.1:6379> ttl 123
(integer) 22
127.0.0.1:6379> ttl 123
(integer) 21
127.0.0.1:6379> ttl 123
(integer) 20
127.0.0.1:6379> ttl 123
(integer) 30
127.0.0.1:6379> ttl 123
(integer) 28
127.0.0.1:6379>
Copier après la connexion
我们再改改代码，看看是否可重入和字段名称是否和我们预期一致
package com.fandf.test.redis;
import org.junit.jupiter.api.Test;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.boot.test.context.SpringBootTest;
import javax.annotation.Resource;
/**
* @Description:
* @author: fandongfeng
* @date: 2023-3-24 16:45
*/
@SpringBootTest
class RedissonTest {
@Resource
private RedissonClient redisson;
@Test
public void watchDog() throws InterruptedException {
RLock lock = redisson.getLock("123");
lock.lock();
lock.lock();
lock.lock();
//加了三次锁，此时重入次数为3
Thread.sleep(3000);
//解锁一次，此时重入次数变为3
lock.unlock();
Thread.sleep(1000000);
}
}Copier après la connexion
127.0.0.1:6379> keys *
1) "123"
127.0.0.1:6379>
127.0.0.1:6379> ttl 123
(integer) 24
127.0.0.1:6379> hgetall 123
1) "df7f4c71-b57b-455f-acee-936ad8475e01:12"
2) "3"
127.0.0.1:6379>
127.0.0.1:6379> hgetall 123
1) "df7f4c71-b57b-455f-acee-936ad8475e01:12"
2) "2"
127.0.0.1:6379>
Copier après la connexion
我们加锁了三次，重入次数是3，字段值也是 uuid+:+threadId,和我们预期结果是一致的。
Redlock算法
redisson是基于Redlock算法实现的，那么什么是Redlock算法呢？
假设当前集群有5个节点，那么运行redlock算法的客户端会一次执行下面步骤
- 1.客户端记录当前系统时间，以毫秒为单位
- 2.依次尝试从5个redis实例中，使用相同key获取锁
当redis请求获取锁时，客户端会设置一个网络连接和响应超时时间，避免因为网络故障等原因导致阻塞。
- 3.客户端使用当前时间减去开始获取锁时间（步骤1的时间），得到获取锁消耗的时间
只有当半数以上redis节点加锁成功，并且加锁消耗的时间要小于锁失效时间，才算锁获取成功
- 4.如果获取到了锁，key的真正有效时间等于锁失效时间 减去 获取锁消耗的时间
- 5.如果获取锁失败，所有的redis实例都会进行解锁
防止因为服务端响应消息丢失，但是实际数据又添加成功导致数据不一致问题
这里有下面几个点需要注意：
- 1.我们都知道单机的redis是cp的，但是集群情况下redis是ap的，所以运行Redisson的节点必须是主节点，不能有从节点，防止主节点加锁成功未同步从节点就宕机，而客户端却收到加锁成功，导致数据不一致问题。
- 2.为了提高redis节点宕机的容错率，可以使用公式2N（n指宕机数量）+1，假设宕机一台，Redisson还要继续运行，那么至少要部署2*1+1=3台主节点。
更多编程相关知识，请访问：编程视频！！
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)
