Einführung in die Implementierungsmethode der verteilten Redis-Sperre

Um verteilte Sperren zu verwenden, müssen mehrere Bedingungen erfüllt sein:

1. Das System ist ein verteiltes System. eigenständig kann mit ReentrantLock oder synchronisierten Codeblöcken implementiert werden)

2. Gemeinsam genutzte Ressourcen (jedes System greift auf dieselbe Ressource zu, der Träger der Ressource kann eine herkömmliche relationale Datenbank oder NoSQL sein)

3, synchroner Zugriff (das heißt, es gibt viele Prozesse, die auf dieselbe gemeinsam genutzte Ressource zugreifen. Wen kümmert es ohne synchronen Zugriff, ob Sie um Ressourcen konkurrieren oder nicht)

2. Beispiele für Anwendungsszenarien

Die Bereitstellungsarchitektur des Management-Backends (mehrere Tomcat-Server + Redis [mehrere Tomcat-Server greifen auf ein Redis zu] + MySQL [mehrere Tomcat-Server greifen auf MySQL auf einem Server zu]) erfüllt die Bedingungen für die Verwendung verteilter Sperren. Mehrere Server müssen auf die Ressourcen des globalen Redis-Cache zugreifen. Wenn keine verteilten Sperren verwendet werden, treten Probleme auf. Schauen Sie sich den folgenden Pseudocode an:

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

Die Hauptfunktionen des obigen Codes sind:

Den Wert N von Redis abrufen, eine Grenzprüfung für den Wert N durchführen, ihn um 1 erhöhen, und dann N zurück an Redis schreiben. Dieses Anwendungsszenario kommt sehr häufig vor, z. B. Flash-Verkäufe, globale inkrementelle ID, IP-Zugriffsbeschränkungen usw.

In Bezug auf IP-Zugriffsbeschränkungen können böswillige Angreifer unbegrenzten Zugriff initiieren, und der Umfang der Parallelität ist relativ groß. Die Grenzprüfung von N in einer verteilten Umgebung ist unzuverlässig, da das Lesen von N von Redis möglicherweise bereits fehlerhaft ist. Daten.

Herkömmliche Sperrmethoden (z. B. Synchronized und Lock von Java) sind nutzlos, da es sich um eine verteilte Umgebung handelt und die Feuerwehrleute, die dieses Synchronisierungsproblem bekämpfen, hilflos sind. In dieser kritischen Phase auf Leben und Tod kommen endlich verteilte Sperren ins Spiel.

Verteilte Sperren können auf viele Arten implementiert werden, z. B. als Zookeeper, Redis ... In beiden Fällen bleibt das Grundprinzip dasselbe: Zur Darstellung der Sperre wird ein Zustandswert verwendet, und die Belegung und Freigabe der Sperre wird durch den Zustandswert identifiziert.

Hier sprechen wir hauptsächlich über die Verwendung von Redis zur Implementierung verteilter Sperren.

3. Verwenden Sie den setNX-Befehl von redis, um verteilte Sperren zu implementieren

1. Implementierungsprinzip

Redis ist ein Einzelprozess-Single-Thread-Modus und verwendet eine Warteschlange. Der Modus verwandelt den gleichzeitigen Zugriff in einen seriellen Zugriff, und es gibt keine Konkurrenz zwischen den Verbindungen mehrerer Clients zu Redis. Der SETNX-Befehl von Redis kann problemlos verteilte Sperren implementieren.

2. Grundlegende Befehlsanalyse

1) setNX (SET if Not eXists)

Syntax:

SETNX key value

Setzen Sie den Wert von key auf value , when Und nur, wenn der Schlüssel nicht existiert.

Wenn der angegebene Schlüssel bereits vorhanden ist, ergreift SETNX keine Aktion.

SETNX ist die Abkürzung für „SET if Not eXists“ (wenn es nicht existiert, dann SET)

Rückgabewert:

Erfolgreich gesetzt, 1 zurückgegeben.

Festlegen fehlgeschlagen, Rückgabe 0.

Beispiel:

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"

Also führen wir den folgenden Befehl aus:

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

Wenn 1 zurückgegeben wird, erhält der Client die Sperre und setzt den Schlüsselwert von lock.foo auf die Zeit Der Wert gibt an, dass der Schlüssel gesperrt ist und der Client die Sperre schließlich über DEL lock.foo aufheben kann.

Wenn 0 zurückgegeben wird, bedeutet dies, dass die Sperre von einem anderen Client erhalten wurde. Zu diesem Zeitpunkt können wir zuerst zurückkehren oder es erneut versuchen und warten, bis die andere Partei den Vorgang abgeschlossen hat, oder warten, bis die Sperre abgelaufen ist.

2) getSET

Syntax:

GETSET key value

Setzen Sie den Wert des angegebenen Schlüssels auf value und geben Sie den alten Wert des Schlüssels zurück.

Wenn der Schlüssel vorhanden ist, aber nicht vom Typ „String“ ist, wird ein Fehler zurückgegeben.

Rückgabewert:

Gibt den alten Wert des angegebenen Schlüssels zurück.

Wenn der Schlüssel keinen alten Wert hat, das heißt, wenn der Schlüssel nicht existiert, wird Null zurückgegeben.

3) get

Syntax:

GET key

Rückgabewert:

Wenn der Schlüssel nicht existiert, geben Sie Null zurück, andernfalls geben Sie den Wert des Schlüssels zurück.

Wenn der Schlüssel kein String-Typ ist, wird ein Fehler zurückgegeben

4. Lösen Sie den Deadlock

Die obige Sperrlogik weist ein Problem auf: Wenn ein Client, der die Sperre hält, ausfällt oder abstürzt und die Sperre nicht aufheben kann.

Anhand des Zeitstempels, der dem Sperrschlüssel entspricht, können wir beurteilen, ob dies geschehen ist. Wenn die aktuelle Zeit größer als der Wert von lock.foo ist, bedeutet dies, dass die Sperre abgelaufen ist und wiederverwendet werden kann.

In diesem Fall können Sie die Sperre nicht einfach über DEL und dann erneut SETNX löschen (aus Gründen der Vernunft sollte der Vorgang zum Löschen der Sperre vom Eigentümer der Sperre durchgeführt werden. Sie müssen dies nur tun Warten Sie, bis die Sperre abgelaufen ist. Wenn mehrere Clients feststellen, dass die Sperre abgelaufen ist, kann es zu einer Race-Bedingung kommen:

C0-Vorgang ist abgelaufen, aber es hält immer noch die Sperre. C1 und C2 haben lock.foo gelesen und den Zeitstempel überprüft und festgestellt, dass sie nacheinander abgelaufen sind.
C1 sendet DEL lock.foo
C1 sendet SETNX lock.foo und es ist erfolgreich.
C2 sendet DEL lock.foo
C2 sendet SETNX lock.foo und es ist erfolgreich.
Auf diese Weise haben sowohl C1 als auch C2 die Sperre! Großes Problem!

Glücklicherweise kann dieses Problem vermieden werden. Schauen wir uns an, wie der C3-Client das macht:

C3 sendet SETNX lock.foo, um Lock zu erhalten, da C0 Hält die Sperre immer noch, gibt Redis eine 0 an C3 zurück

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入门教程栏目。

Das obige ist der detaillierte Inhalt vonEinführung in die Implementierungsmethode der verteilten Redis-Sperre. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!