Introduction to the implementation method of redis distributed lock

1. Several conditions must be met to use distributed locks:

1. The system is a distributed system (the key is distributed , a single machine can be implemented using ReentrantLock or synchronized code blocks)

2. Shared resources (each system accesses the same resource, and the carrier of the resource may be a traditional relational database or NoSQL)

3 , Synchronous access (that is, there are many processes accessing the same shared resource at the same time. Without synchronous access, who cares whether you compete for resources or not)

2. Application scenario examples

The deployment architecture of the management background (multiple tomcat servers redis [multiple tomcat servers access one redis] mysql [multiple tomcat servers access mysql on one server]) meets the conditions for using distributed locks. Multiple servers need to access the resources of the redis global cache. Problems will occur if distributed locks are not used. Look at the following pseudo code:

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

The main functions of the above code are:

Get the value N from redis, perform a boundary check on the value N, increment it by 1, and then write N back to redis. This application scenario is very common, such as flash sales, global incremental ID, IP access restrictions, etc.

In terms of IP access restrictions, malicious attackers may initiate unlimited access, and the amount of concurrency is relatively large. The boundary check of N is unreliable in a distributed environment, because N read from redis may already be dirty. data.

Traditional locking methods (such as Java's synchronized and Lock) are useless, because this is a distributed environment, and the firefighters who are fighting this synchronization problem are helpless. In this critical period of life and death, distributed locks finally come into play.

Distributed locks can be implemented in many ways, such as zookeeper, redis.... Either way, the basic principle remains the same: a state value is used to represent the lock, and the occupation and release of the lock are identified by the state value.

Here we mainly talk about how to use redis to implement distributed locks.

3. Use the setNX command of redis to implement distributed locks

1. Implementation principle

Redis is a single-process single-thread mode and uses a queue The mode turns concurrent access into serial access, and there is no competition between multiple clients' connections to Redis. The SETNX command of redis can easily implement distributed locks.

2. Basic command analysis

1) setNX (SET if Not eXists)

Syntax:

SETNX key value

Set the value of key to value, when And only if key does not exist.

If the given key already exists, SETNX does not take any action.

SETNX is the abbreviation of "SET if Not eXists" (if it does not exist, then SET)

Return value:

Set successfully, return 1.

Failed to set, return 0.

Example:

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"

So we execute the following command

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

If 1 is returned, the client obtains the lock and sets the key value of lock.foo to the time The value indicates that the key is locked and the client can finally release the lock via DEL lock.foo.

If 0 is returned, it means that the lock has been obtained by another client. At this time, we can return first or retry and wait for the other party to complete or wait for the lock to time out.

2) getSET

Syntax:

GETSET key value

Set the value of the given key to value and return the old value of key.

When key exists but is not of string type, an error is returned.

Return value:

Return the old value of the given key.

When key has no old value, that is, when key does not exist, nil is returned.

3) get

Syntax:

GET key

Return value:

When key does not exist, return nil, otherwise, return the value of key.

If key is not a string type, then an error is returned

4. Solve the deadlock

The above locking logic has a problem: if a The client holding the lock fails or crashes and cannot release the lock. How to solve the problem?

We can judge whether this has happened through the timestamp corresponding to the lock key. If the current time is greater than the value of lock.foo, it means that the lock has expired and can be reused.

When this happens, you cannot simply delete the lock through DEL and then SETNX again (to be reasonable, the operation of deleting the lock should be performed by the owner of the lock. You only need to wait for it here. timeout), when multiple clients detect that the lock has timed out, they will try to release it. A race condition may occur here. Let us simulate this scenario:

C0 operation has timed out. , but it still holds the lock. C1 and C2 read lock.foo and checked the timestamp, and found that they timed out one after another.
C1 sends DEL lock.foo
C1 sends SETNX lock.foo and succeeds.
C2 sends DEL lock.foo
C2 sends SETNX lock.foo and succeeds.
In this way, C1 and C2 both get the lock! Big problem!

Fortunately, this problem can be avoided. Let's take a look at how the C3 client does it:

C3 sends SETNX lock.foo to obtain Lock, since C0 still holds the lock, Redis returns a 0 to 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入门教程栏目。

The above is the detailed content of Introduction to the implementation method of redis distributed lock. For more information, please follow other related articles on the PHP Chinese website!