Redis实现分布式锁

一个简单的用户操作，一个线程去修改用户信息，首先从数据库中读取用户的信息，然后在内存中修改，然后存回去。
单线程中，这个操作是没问题的。但是在多线程的环境中，读取，修改，存储是三个操作，不是原子操作，所以在多线程中，这样会发生线程不安全的问题。
对于这种问题，我们可以使用分布式锁来让程序同步执行。
分布式锁的思路：一个线程先占用资源，另外的线程无法访问，会阻塞或者稍后请求。

edis可以使用string类型的setnx key value指令来实现分布式锁，这条指令只会在key值不存在时才进行set的操作，由于redis本身的单线程模型，所有的指令都是同步执行的，所以非常适合解决分布式锁这个问题。
但是一个成熟的分布式锁，需要考虑以下问题：

加锁与释放锁的过程中如果程序发生了异常，导致没有执行释放锁的操作，那么当前线程将永远持有这把锁，而其他线程则无法访问该资源，处于阻塞状态，造成死锁。

为了解决上述问题，可以给key设置一个过期时间，那么key会在一段时间之后自动过期，其他线程得到锁，就可以正常轮转了。可以通过Redis 2.8之后的API来实现：

SET resource_name my_random_value NX PX 30000

这条指令将setnx和设置过期时间结合到了一起，具备原子性。
但是尚未解决超时时间带来的问题.

超时时间和宕机带来的问题

此时还可能会发生几种事情，如果设置的时间过短（假设为5S），A线程（需要执行8S）在过期时间的范围内并未完全执行完代码，过了规定的时间后，B线程获取了这把锁，然后3S之后，A线程执行完了代码，开始释放资源，那么此时B线程的锁就会被A线程所释放了，此时会造成业务发生混乱。
宕机的情况，此处引入一下Redis中文网站的的描述.

为了不发生这种灾难，我们还需要借助LUA脚本提高释放锁的容错性。
解决方案是：为每个线程分配随机数，在释放锁的时候，先对比value值是否相同，如果不相同，则不用释放（key会自动过期）。相同的时候，进行释放.

if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end

下面贴一下完整的Java代码。

import redis.clients.jedis.params.SetParams;

import java.util.Arrays;
import java.util.UUID;

/**
 * 实践redis的分布式锁
 */
public class distributed_lock {
    public static void main(String[] args) {
        Runnable runnable = new Runnable(){
            @Override
            public void run(){
                System.out.println(Thread.currentThread().getName());
                disributedLock();
            }
        };
        Thread thread = new Thread(runnable);
        Thread thread1 = new Thread(runnable);
        thread.start();
        thread1.start();

    }

    static void  disributedLock(){
        Redis redis = new Redis();
        redis.exeute(jedis -> {
            String uuid = UUID.randomUUID().toString();
            String result = jedis.set("k1", uuid, new SetParams().nx().ex(5));
            if(result!=null && "OK".equals(result)){
                /**
                 * 如果这里的代码出现异常，会导致资源（锁）无法释放，导致其他线程无法得到该资源。
                 * 可以设置过期时间，让Key在一段时间之后自动过期
                 * 设置过期时间，也会存在问题，如果服务器在获取锁和设置过期时间之间挂掉了，那么锁还是无法被释放
                 * 也会造成死锁，因为这是两个操作，不具备原子性。
                 * 在Redis 2.8的版本中，Redis发布了一个新指令,value最好设置成随机数，官网推荐
                 * value的值必须是随机数主要是为了更安全的释放锁，释放锁的时候使用脚本告诉Redis:
                 * 只有key存在并且存储的值和我指定的值一样才能告诉我删除成功
                 * if redis.call("get",KEYS[1]) == ARGV[1] then
                 *     return redis.call("del",KEYS[1])
                 * else
                 *     return 0
                 * end
                 * ---------------------------
                 * set key value nx ex/px time
                 * 关于超时时间的问题：
                 * 如果执行的业务消耗的时间不一致，可能会出现凌乱。
                 * A线程执行了8S，B线程执行了5秒，那么在B执行的过程中，A可能会释放掉Key,让锁失效。
                 * value设置为随机数的话，可以比较value再释放资源.否则不释放
                 * ------------------通过Lua脚本缓存比较value的这个操作，它是原子性的
                 * cat lua/releaseWhereValueEquals.lua | redis-cli -a 123 script load --pipe
                 * -----------SHA1校验码
                 * b8059ba43af6ffe8bed3db65bac35d452f8115d8
                 */
                jedis.set("hello", "world");//没人占位
                System.out.println(jedis.get("hello"));
                //解铃还需系铃人,释放自己的锁
                jedis.evalsha("b8059ba43af6ffe8bed3db65bac35d452f8115d8", Arrays.asList("k1"),Arrays.asList(uuid));
            }else {
                //有人占位,停止/暂缓操作
                System.out.println("先等等...");
            }
        });
    }
}

原文出处