Redis事务锁

在不同进程需要互斥地访问共享资源时，分布式锁是一种非常有用的技术手段。本文采用Spring Data Redis实现一下Redis的分布式事务锁。

Redis为单进程单线程模式，采用队列模式将并发访问变成串行访问，且多客户端对Redis的连接并不存在竞争关系。

SETNX命令（SET if Not eXists）语法：

SETNX key value

若给定的 key 已经存在，则 SETNX 不做任何动作，并返回0。

安全性：保证互斥，在任何时候，只有一个客户端可以持有锁
无死锁：即使当前持有锁的客户端崩溃或者从集群中被分开了，其它客户端最终总是能够获得锁。
容错性：只要大部分的 Redis 节点在线，那么客户端就能够获取和释放锁。

使用Spring redisTemplate的实现

使用redisTemplate实现需要配合redis 的eval实现，在Spring Data Redis的官方文档中Redis Scripting一节有相关的说明。

先看一下Spring Redis文档中是如何使用eval的：

@Beanpublic RedisScript<Boolean> script() {  DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<Boolean>();  redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("META-INF/scripts/checkandset.lua")));  redisScript.setResultType(Boolean.class);}1
2
3
4
5
6

public class Example {  @Autowired  RedisScript<Boolean> script;  public boolean checkAndSet(String expectedValue, String newValue) {    return redisTemplate.execute(script, Collections.singletonList("key"), expectedValue, newValue);  }}1
2
3
4
5
6
7

 -- checkandset.lua local current = redis.call('GET', KEYS[1]) if current == ARGV[1]   then redis.call('SET', KEYS[1], ARGV[2])   return true end return false1
2
3
4
5
6
7

关于eval函数以及Lua脚本在此不进行赘述，下面来看一下我们如何使用redisTemplate实现事务锁。

定义事务锁的Bean：

public class RedisLock {    private String key;    private final UUID uuid;    private long lockTimeout;    private long startLockTimeMillis;    private long getLockTimeMillis;    private int tryCount;    public RedisLock(String key, UUID uuid, long lockTimeout, long startLockTimeMillis, long getLockTimeMillis, int tryCount) {        this.key = key;        this.uuid = uuid;        this.lockTimeout = lockTimeout;        this.startLockTimeMillis = startLockTimeMillis;        this.getLockTimeMillis = getLockTimeMillis;        this.tryCount = tryCount;    }    public String getKey() {        return key;    }    public void setKey(String key) {        this.key = key;    }    public UUID getUuid() {        return uuid;    }    public long getLockTimeout() {        return lockTimeout;    }    public void setLockTimeout(long lockTimeout) {        this.lockTimeout = lockTimeout;    }    public long getGetLockTimeMillis() {        return getLockTimeMillis;    }    public void setGetLockTimeMillis(long getLockTimeMillis) {        this.getLockTimeMillis = getLockTimeMillis;    }    public long getStartLockTimeMillis() {        return startLockTimeMillis;    }    public void setStartLockTimeMillis(long startLockTimeMillis) {        this.startLockTimeMillis = startLockTimeMillis;    }    public int getTryCount() {        return tryCount;    }    public void setTryCount(int tryCount) {        this.tryCount = tryCount;    }}1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

创建获取锁操作：

// 锁的过期时间，单位毫秒private static final long DEFAULT_LOCK_TIME_OUT = 3000;   // 争抢锁的超时时间，单位毫秒，0代表永不超时（一直抢到锁为止）private static final long DEFAULT_TRY_LOCK_TIME_OUT = 0;  //拿锁的EVAL函数private static final String LUA_SCRIPT_LOCK = "return redis.call('SET', KEYS[1], ARGV[1], 'NX', 'PX', ARGV[2]) ";//释放锁的EVAL函数private static RedisScript<String> scriptLock = new DefaultRedisScript<String>(LUA_SCRIPT_LOCK, String.class);1
2
3
4
5
6
7

获取锁的方法：

public static RedisLock lock(int dbIndex, String key, long lockTimeout, long tryLockTimeout) {    long timestamp = System.currentTimeMillis();    try {        //锁的名称        key = key + ".lock";        UUID uuid = UUID.randomUUID();        int tryCount = 0;        //在超时之前，循环尝试拿锁        while (tryLockTimeout == 0 || (System.currentTimeMillis() - timestamp) < tryLockTimeout) {//执行拿锁的操作，注意这里，后面的三个参数分别对应了scriptLock字符串中的三个变量值，KEYS[1]，ARGV[1]，ARGV[2]，含义为锁的key，key对应的value，以及key 的存在时间(单位毫秒)String result = redisTemplate.execute(scriptLock, redisTemplate.getStringSerializer(), redisTemplate.getStringSerializer(), Collections.singletonList(key), uuid.toString(),                        String.valueOf(lockTimeout));    tryCount++;    //返回“OK”代表拿到锁    if (result != null && result.equals("OK")) {        return new RedisLock(key, uuid, lockTimeout, timestamp, System.currentTimeMillis(), tryCount);    } else {        try {            //如果失败，休息50毫秒继续重试（自旋锁）            Thread.sleep(50);        } catch (InterruptedException e) {                        e.printStackTrace();        }    }}logger.error("Fail to get lock key");}return null;}1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

上述代码就是通过redisTemplate实现的redis 的分布式锁，如果创建Bean成功则说明拿到锁，否则拿锁失败，核心是采用Redis 的eval函数，使用类似CAS的操作，进行拿锁，如果拿锁成功，则返回“OK”，如果失败，休眠然后继续尝试拿锁，直到超时。

释放锁操作：

private static final String LUA_SCRIPT_UNLOCK =             "if (redis.call('GET', KEYS[1]) == ARGV[1]) then "            + "return redis.call('DEL',KEYS[1]) "             + "else " + "return 0 " + "end";private static RedisScript<String> scriptUnlock =         new DefaultRedisScript<String>(LUA_SCRIPT_UNLOCK,            String.class);1
2
3
4
5
6
7
8
9

public static void unLock(int dbIndex, RedisLock lock) {                                                                     redisTemplate.execute(scriptUnlock,                  redisTemplate.getStringSerializer(),                  redisTemplate.getStringSerializer(),                       Collections.singletonList(lock.getKey()),                       lock.getUuid().toString());}1
2
3
4
5
6
7
8

上述就是使用Redis来实现分布式锁，其方法是采用Redis String 的 SET进行实现，SET 命令的行为可以通过一系列参数来修改：

EX second ：设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value 。
PX millisecond ：设置键的过期时间为 millisecond 毫秒。 SET key value PX millisecond 效果等同于 PSETEX key millisecond value 。
NX ：只在键不存在时，才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value 。
XX ：只在键已经存在时，才对键进行设置操作。

本站仅提供存储服务，所有内容均由用户发布，如发现有害或侵权内容，请点击举报。