Introduction à la méthode d'implémentation du verrouillage distribué Redis

1. Plusieurs conditions doivent être remplies pour utiliser les serrures distribuées :

1. Le système est un système distribué (la clé est distribuée, autonome peut être implémenté à l'aide de ReentrantLock ou de blocs de code synchronisés)

2 Ressources partagées (chaque système accède à la même ressource, le support de la ressource peut être une base de données relationnelle traditionnelle ou NoSQL)

3, accès synchrone (c'est-à-dire qu'il existe de nombreux processus accédant à la même ressource partagée. Sans accès synchrone, peu importe que vous soyez en compétition pour les ressources ou non)

2. Exemples de scénarios d'application

L'architecture de déploiement du backend de gestion (plusieurs serveurs Tomcat + redis [plusieurs serveurs Tomcat accèdent à un redis] + mysql [plusieurs serveurs Tomcat accèdent à mysql sur un serveur]) répond aux conditions d'utilisation des verrous distribués. Plusieurs serveurs doivent accéder aux ressources du cache global Redis. Des problèmes se produiront si les verrous distribués ne sont pas utilisés. Regardez le pseudo code suivant :

long N=0L;
//N从redis获取值
if(N<5){
N++；
//N写回redis
}

Les principales fonctions du code ci-dessus sont :

Récupérez la valeur N de redis, effectuez une vérification des limites sur la valeur N, incrémentez-la de 1, puis réécrivez N dans Redis. Ce scénario d'application est très courant, comme les ventes flash, l'identification incrémentielle globale, les restrictions d'accès IP, etc.

En termes de restrictions d'accès IP, des attaquants malveillants peuvent lancer un accès illimité et le degré de concurrence est relativement important. La vérification des limites de N dans un environnement distribué n'est pas fiable car N lu à partir de Redis peut déjà être sale. données.

Les méthodes de verrouillage traditionnelles (telles que Synchronized et Lock de Java) sont inutiles, car il s'agit d'un environnement distribué, et les pompiers qui luttent contre ce problème de synchronisation sont impuissants. Dans cette période critique de vie ou de mort, les verrous distribués entrent enfin en jeu.

Les verrous distribués peuvent être implémentés de plusieurs manières, telles que zookeeper, redis... Quoi qu'il en soit, le principe de base reste inchangé : une valeur d'état est utilisée pour représenter la serrure, et l'occupation et la libération de la serrure sont identifiées par la valeur d'état.

Ici, nous parlons principalement de la façon d'utiliser Redis pour implémenter des verrous distribués.

3. Utilisez la commande setNX de redis pour implémenter des verrous distribués

1. Principe de mise en œuvre

Redis est un mode mono-thread à processus unique. et utilise une file d'attente. Le mode transforme l'accès simultané en accès série et il n'y a pas de concurrence entre les connexions de plusieurs clients à Redis. La commande SETNX de redis peut facilement implémenter des verrous distribués.

2. Analyse des commandes de base

1) setNX (SET if Not eXists)

Syntaxe :

SETNX key value

Définissez la valeur de key sur value , lorsque Et seulement si la clé n'existe pas.

Si la clé donnée existe déjà, SETNX n'effectue aucune action.

SETNX est l'abréviation de "SET if Not eXists" (s'il n'existe pas, alors SET)

Valeur de retour :

Défini avec succès, renvoie 1.

Échec de la définition, renvoyant 0.

Exemple :

redis> EXISTS job                # job 不存在
(integer) 0

redis> SETNX job "programmer"    # job 设置成功
(integer) 1

redis> SETNX job "code-farmer"   # 尝试覆盖 job ，失败
(integer) 0

redis> GET job                   # 没有被覆盖
"programmer"

On exécute donc la commande suivante

SETNX lock.foo <current Unix time + lock timeout + 1>

Si 1 est renvoyé, le client obtient le verrou et définit la valeur clé de lock.foo à l'heure La valeur indique que la clé est verrouillée et que le client peut enfin libérer le verrou via DEL lock.foo.

Si 0 est renvoyé, cela signifie que le verrouillage a été obtenu par un autre client. À ce moment, nous pouvons revenir en premier ou réessayer et attendre que l'autre partie termine ou attendre que le verrouillage expire.

2) getSET

Syntaxe :

GETSET key value

Définissez la valeur de la clé donnée sur value et renvoyez l'ancienne valeur de la clé.

Lorsque la clé existe mais n'est pas de type chaîne, renvoie une erreur.

Valeur de retour :

Renvoie l'ancienne valeur de la clé donnée.

Lorsque la clé n'a pas d'ancienne valeur, c'est-à-dire lorsque la clé n'existe pas, nil est renvoyé.

3) get

Syntaxe :

GET key

Valeur de retour :

Lorsque la clé n'existe pas, renvoie nil, sinon, renvoie la valeur de la clé.

Si la clé n'est pas un type chaîne, une erreur est renvoyée

4. Résoudre l'impasse

La logique de verrouillage ci-dessus a un problème : si un Le client détenant le verrou échoue ou plante et ne peut pas libérer le verrou. Comment résoudre le problème ?

Nous pouvons juger si cela s'est produit grâce à l'horodatage correspondant à la clé de verrouillage. Si l'heure actuelle est supérieure à la valeur de lock.foo, cela signifie que le verrou a expiré et peut être réutilisé.

Lorsque cela se produit, vous ne pouvez pas simplement supprimer le verrou via DEL puis à nouveau SETNX (raisonnablement, l'opération de suppression du verrou doit être effectuée par le propriétaire du verrou. Il vous suffit d'attendre ici. timeout), lorsque plusieurs clients détectent que le verrou a expiré, ils essaieront de le libérer. Une condition de concurrence peut se produire ici :

L'opération C0 a été chronométrée. out. , mais il détient toujours le verrou C1 et C2, lisez lock.foo et vérifiez l'horodatage et constatez qu'ils ont expiré l'un après l'autre.

C1 envoie DEL lock.foo
C1 envoie SETNX lock.foo et il réussit.
C2 envoie DEL lock.foo
C2 envoie SETNX lock.foo et il réussit.
De cette façon, C1 et C2 ont tous deux le verrou ! Gros problème !

Heureusement, ce problème peut être évité. Voyons comment le client C3 procède :

C3 envoie SETNX lock.foo pour obtenir Lock, depuis C0. tient toujours le verrou, Redis renvoie un 0 à C3

C3发送GET lock.foo 以检查锁是否超时了，如果没超时，则等待或重试。

反之，如果已超时，C3通过下面的操作来尝试获得锁：

GETSET lock.foo

通过GETSET，C3拿到的时间戳如果仍然是超时的，那就说明，C3如愿以偿拿到锁了。

如果在C3之前，有个叫C4的客户端比C3快一步执行了上面的操作，那么C3拿到的时间戳是个未超时的值，这时，C3没有如期获得锁，需要再次等待或重试。留意一下，尽管C3没拿到锁，但它改写了C4设置的锁的超时值，不过这一点非常微小的误差带来的影响可以忽略不计。

注意：为了让分布式锁的算法更稳键些，持有锁的客户端在解锁之前应该再检查一次自己的锁是否已经超时，再去做DEL操作，因为可能客户端因为某个耗时的操作而挂起，操作完的时候锁因为超时已经被别人获得，这时就不必解锁了。

五、代码实现

expireMsecs 锁持有超时，防止线程在入锁以后，无限的执行下去，让锁无法释放
timeoutMsecs 锁等待超时，防止线程饥饿，永远没有入锁执行代码的机会

注意：项目里面需要先搭建好redis的相关配置

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

/**
 * Redis distributed lock implementation.
 *
 * @author zhengcanrui
 */
public class RedisLock {

    private static Logger logger = LoggerFactory.getLogger(RedisLock.class);

    private RedisTemplate redisTemplate;

    private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 100;

    /**
     * Lock key path.
     */
    private String lockKey;

    /**
     * 锁超时时间，防止线程在入锁以后，无限的执行等待
     */
    private int expireMsecs = 60 * 1000;

    /**
     * 锁等待时间，防止线程饥饿
     */
    private int timeoutMsecs = 10 * 1000;

    private volatile boolean locked = false;

    /**
     * Detailed constructor with default acquire timeout 10000 msecs and lock expiration of 60000 msecs.
     *
     * @param lockKey lock key (ex. account:1, ...)
     */
    public RedisLock(RedisTemplate redisTemplate, String lockKey) {
        this.redisTemplate = redisTemplate;
        this.lockKey = lockKey + "_lock";
    }

    /**
     * Detailed constructor with default lock expiration of 60000 msecs.
     *
     */
    public RedisLock(RedisTemplate redisTemplate, String lockKey, int timeoutMsecs) {
        this(redisTemplate, lockKey);
        this.timeoutMsecs = timeoutMsecs;
    }

    /**
     * Detailed constructor.
     *
     */
    public RedisLock(RedisTemplate redisTemplate, String lockKey, int timeoutMsecs, int expireMsecs) {
        this(redisTemplate, lockKey, timeoutMsecs);
        this.expireMsecs = expireMsecs;
    }

    /**
     * @return lock key
     */
    public String getLockKey() {
        return lockKey;
    }

    private String get(final String key) {
        Object obj = null;
        try {
            obj = redisTemplate.execute(new RedisCallback<Object>() {
                @Override
                public Object doInRedis(RedisConnection connection) throws DataAccessException {
                    StringRedisSerializer serializer = new StringRedisSerializer();
                    byte[] data = connection.get(serializer.serialize(key));
                    connection.close();
                    if (data == null) {
                        return null;
                    }
                    return serializer.deserialize(data);
                }
            });
        } catch (Exception e) {
            logger.error("get redis error, key : {}", key);
        }
        return obj != null ? obj.toString() : null;
    }

    private boolean setNX(final String key, final String value) {
        Object obj = null;
        try {
            obj = redisTemplate.execute(new RedisCallback<Object>() {
                @Override
                public Object doInRedis(RedisConnection connection) throws DataAccessException {
                    StringRedisSerializer serializer = new StringRedisSerializer();
                    Boolean success = connection.setNX(serializer.serialize(key), serializer.serialize(value));
                    connection.close();
                    return success;
                }
            });
        } catch (Exception e) {
            logger.error("setNX redis error, key : {}", key);
        }
        return obj != null ? (Boolean) obj : false;
    }

    private String getSet(final String key, final String value) {
        Object obj = null;
        try {
            obj = redisTemplate.execute(new RedisCallback<Object>() {
                @Override
                public Object doInRedis(RedisConnection connection) throws DataAccessException {
                    StringRedisSerializer serializer = new StringRedisSerializer();
                    byte[] ret = connection.getSet(serializer.serialize(key), serializer.serialize(value));
                    connection.close();
                    return serializer.deserialize(ret);
                }
            });
        } catch (Exception e) {
            logger.error("setNX redis error, key : {}", key);
        }
        return obj != null ? (String) obj : null;
    }

    /**
     * 获得 lock.
     * 实现思路: 主要是使用了redis 的setnx命令,缓存了锁.
     * reids缓存的key是锁的key,所有的共享, value是锁的到期时间(注意:这里把过期时间放在value了,没有时间上设置其超时时间)
     * 执行过程:
     * 1.通过setnx尝试设置某个key的值,成功(当前没有这个锁)则返回,成功获得锁
     * 2.锁已经存在则获取锁的到期时间,和当前时间比较,超时的话,则设置新的值
     *
     * @return true if lock is acquired, false acquire timeouted
     * @throws InterruptedException in case of thread interruption
     */
    public synchronized boolean lock() throws InterruptedException {
        int timeout = timeoutMsecs;
        while (timeout >= 0) {
            long expires = System.currentTimeMillis() + expireMsecs + 1;
            String expiresStr = String.valueOf(expires); //锁到期时间
            if (this.setNX(lockKey, expiresStr)) {
                // lock acquired
                locked = true;
                return true;
            }

            String currentValueStr = this.get(lockKey); //redis里的时间
            if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
                //判断是否为空，不为空的情况下，如果被其他线程设置了值，则第二个条件判断是过不去的
                // lock is expired

                String oldValueStr = this.getSet(lockKey, expiresStr);
                //获取上一个锁到期时间，并设置现在的锁到期时间，
                //只有一个线程才能获取上一个线上的设置时间，因为jedis.getSet是同步的
                if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
                    //防止误删（覆盖，因为key是相同的）了他人的锁——这里达不到效果，这里值会被覆盖，但是因为什么相差了很少的时间，所以可以接受

                    //[分布式的情况下]:如过这个时候，多个线程恰好都到了这里，但是只有一个线程的设置值和当前值相同，他才有权利获取锁
                    // lock acquired
                    locked = true;
                    return true;
                }
            }
            timeout -= DEFAULT_ACQUIRY_RESOLUTION_MILLIS;

            /*
                延迟100 毫秒,  这里使用随机时间可能会好一点,可以防止饥饿进程的出现,即,当同时到达多个进程,
                只会有一个进程获得锁,其他的都用同样的频率进行尝试,后面有来了一些进行,也以同样的频率申请锁,这将可能导致前面来的锁得不到满足.
                使用随机的等待时间可以一定程度上保证公平性
             */
            Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);

        }
        return false;
    }


    /**
     * Acqurired lock release.
     */
    public synchronized void unlock() {
        if (locked) {
            redisTemplate.delete(lockKey);
            locked = false;
        }
    }

}

调用：

RedisLock lock = new RedisLock(redisTemplate, key, 10000, 20000);
 try {
            if(lock.lock()) {
                   //需要加锁的代码
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            //为了让分布式锁的算法更稳键些，持有锁的客户端在解锁之前应该再检查一次自己的锁是否已经超时，再去做DEL操作，因为可能客户端因为某个耗时的操作而挂起，
            //操作完的时候锁因为超时已经被别人获得，这时就不必解锁了。 ————这里没有做
            lock.unlock();
        }

六、一些问题

1、为什么不直接使用expire设置超时时间，而将时间的毫秒数其作为value放在redis中？

如下面的方式，把超时的交给redis处理：

lock(key, expireSec){
isSuccess = setnx key
if (isSuccess)
expire key expireSec
}

这种方式貌似没什么问题，但是假如在setnx后，redis崩溃了，expire就没有执行，结果就是死锁了。锁永远不会超时。

 2、为什么前面的锁已经超时了，还要用getSet去设置新的时间戳的时间获取旧的值，然后和外面的判断超时时间的时间戳比较呢？

因为是分布式的环境下，可以在前一个锁失效的时候，有两个进程进入到锁超时的判断。如：

C0超时了，还持有锁,C1/C2同时请求进入了方法里面

C1/C2获取到了C0的超时时间

C1使用getSet方法

C2也执行了getSet方法

假如我们不加 oldValueStr.equals(currentValueStr) 的判断，将会C1/C2都将获得锁，加了之后，能保证C1和C2只能一个能获得锁，一个只能继续等待。

注意：这里可能导致超时时间不是其原本的超时时间，C1的超时时间可能被C2覆盖了，但是他们相差的毫秒及其小，这里忽略了。

更多redis知识请关注redis入门教程栏目。

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!