Redisson分佈式鎖實現

1.  基本用法

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.8.2</version>
</dependency>

Config config = new Config();
config.useClusterServers()
    .setScanInterval(2000) // cluster state scan interval in milliseconds
    .addNodeAddress("redis://127.0.0.1:7000", "redis://127.0.0.1:7001")
    .addNodeAddress("redis://127.0.0.1:7002");

RedissonClient redisson = Redisson.create(config);

RLock lock = redisson.getLock("anyLock");

lock.lock();

try {
    ...
} finally {
    lock.unlock();
}

針對上面這段代碼，重點看一下Redisson是如何基於Redis實現分佈式鎖的

Redisson中提供的加鎖的方法有不少，但大體相似，此處只看lock()方法

更多請參見 https://github.com/redisson/redisson/wiki/8.-distributed-locks-and-synchronizers

2.  加鎖

能夠看到，調用getLock()方法後實際返回一個RedissonLock對象，在RedissonLock對象的lock()方法主要調用tryAcquire()方法

因爲leaseTime == -1，因而走tryLockInnerAsync()方法，這個方法纔是關鍵

首先，看一下evalWriteAsync方法的定義

<T, R> RFuture<R> evalWriteAsync(String key, Codec codec, RedisCommand<T> evalCommandType, String script, List<Object> keys, Object ... params);

最後兩個參數分別是keys和params

實際調用是這樣的：

單獨將調用的那一段摘出來看

commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                  "if (redis.call('exists', KEYS[1]) == 0) then " +
                      "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                      "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "return redis.call('pttl', KEYS[1]);",
                    Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));

結合上面的參數聲明，咱們能夠知道，這裏KEYS[1]就是getName()，ARGV[2]是getLockName(threadId)

假設前面獲取鎖時傳的name是「abc」，假設調用的線程ID是Thread-1，假設成員變量UUID類型的id是6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c

那麼KEYS[1]=abc，ARGV[2]=6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1

所以，這段腳本的意思是

　　一、判斷有沒有一個叫「abc」的key

　　二、若是沒有，則在其下設置一個字段爲「6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1」，值爲「1」的鍵值對 ，並設置它的過時時間

　　三、若是存在，則進一步判斷「6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1」是否存在，若存在，則其值加1，並從新設置過時時間

　　四、返回「abc」的生存時間（毫秒）

這裏用的數據結構是hash，hash的結構是： key  字段1  值1 字段2  值2  。。。

用在鎖這個場景下，key就表示鎖的名稱，也能夠理解爲臨界資源，字段就表示當前得到鎖的線程

全部競爭這把鎖的線程都要判斷在這個key下有沒有本身線程的字段，若是沒有則不能得到鎖，若是有，則至關於重入，字段值加1（次數）

3.  解鎖

protected RFuture<Boolean> unlockInnerAsync(long threadId) {
    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            "if (redis.call('exists', KEYS[1]) == 0) then " +
                "redis.call('publish', KEYS[2], ARGV[1]); " +
                "return 1; " +
            "end;" +
            "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                "return nil;" +
            "end; " +
            "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
            "if (counter > 0) then " +
                "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                "return 0; " +
            "else " +
                "redis.call('del', KEYS[1]); " +
                "redis.call('publish', KEYS[2], ARGV[1]); " +
                "return 1; "+
            "end; " +
            "return nil;",
            Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));

}

咱們仍是假設name=abc，假設線程ID是Thread-1

同理，咱們能夠知道

KEYS[1]是getName()，即KEYS[1]=abc

KEYS[2]是getChannelName()，即KEYS[2]=redisson_lock__channel:{abc}

ARGV[1]是LockPubSub.unlockMessage，即ARGV[1]=0

ARGV[2]是生存時間

ARGV[3]是getLockName(threadId)，即ARGV[3]=6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1

所以，上面腳本的意思是：

　　一、判斷是否存在一個叫「abc」的key

　　二、若是不存在，向Channel中廣播一條消息，廣播的內容是0，並返回1

　　三、若是存在，進一步判斷字段6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1是否存在

　　四、若字段不存在，返回空，若字段存在，則字段值減1

　　五、若減完之後，字段值仍大於0，則返回0

　　六、減完後，若字段值小於或等於0，則廣播一條消息，廣播內容是0，並返回1；

能夠猜想，廣播0表示資源可用，即通知那些等待獲取鎖的線程如今能夠得到鎖了

4.  等待

以上是正常狀況下獲取到鎖的狀況，那麼當沒法當即獲取到鎖的時候怎麼辦呢？

再回到前面獲取鎖的位置

@Override
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
    long threadId = Thread.currentThread().getId();
    Long ttl = tryAcquire(leaseTime, unit, threadId);
    // lock acquired
    if (ttl == null) {
        return;
    }

    //    訂閱
    RFuture<RedissonLockEntry> future = subscribe(threadId);
    commandExecutor.syncSubscription(future);

    try {
        while (true) {
            ttl = tryAcquire(leaseTime, unit, threadId);
            // lock acquired
            if (ttl == null) {
                break;
            }

            // waiting for message
            if (ttl >= 0) {
                getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
            } else {
                getEntry(threadId).getLatch().acquire();
            }
        }
    } finally {
        unsubscribe(future, threadId);
    }
//        get(lockAsync(leaseTime, unit));
}


protected static final LockPubSub PUBSUB = new LockPubSub();

protected RFuture<RedissonLockEntry> subscribe(long threadId) {
    return PUBSUB.subscribe(getEntryName(), getChannelName(), commandExecutor.getConnectionManager().getSubscribeService());
}

protected void unsubscribe(RFuture<RedissonLockEntry> future, long threadId) {
    PUBSUB.unsubscribe(future.getNow(), getEntryName(), getChannelName(), commandExecutor.getConnectionManager().getSubscribeService());
}

這裏會訂閱Channel，當資源可用時能夠及時知道，並搶佔，防止無效的輪詢而浪費資源

當資源可用用的時候，循環去嘗試獲取鎖，因爲多個線程同時去競爭資源，因此這裏用了信號量，對於同一個資源只容許一個線程得到鎖，其它的線程阻塞

5.  小結

 

 

 

6.  其它相關

《基於Redis的分佈式鎖的簡單實現》