快捷導(dǎo)航

Redis Redisson lock和tryLock的原理分析

更新時間：2024年04月23日 09:39:34 作者：奧特迦

這篇文章主要介紹了Redis Redisson lock和tryLock的原理分析，具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

1. 工具類

package com.meta.mall.common.utils;

import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * redisson 分布式工具類
 *
 * @author gaoyang
 * @date 2022-05-14 08:58
 */
@Slf4j
@Component
public class RedissonUtils {

    @Resource
    private RedissonClient redissonClient;

    /**
     * 加鎖
     *
     * @param lockKey
     */
    public void lock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock();
    }

    /**
     * 帶過期時間的鎖
     *
     * @param lockKey   key
     * @param leaseTime 上鎖后自動釋放鎖時間
     */
    public void lock(String lockKey, long leaseTime) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(leaseTime, TimeUnit.SECONDS);
    }

    /**
     * 帶超時時間的鎖
     *
     * @param lockKey   key
     * @param leaseTime 上鎖后自動釋放鎖時間
     * @param unit      時間單位
     */
    public void lock(String lockKey, long leaseTime, TimeUnit unit) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(leaseTime, unit);
    }

    /**
     * 嘗試獲取鎖
     *
     * @param lockKey key
     * @return
     */
    public boolean tryLock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        return lock.tryLock();
    }

    /**
     * 嘗試獲取鎖
     *
     * @param lockKey   key
     * @param waitTime  最多等待時間
     * @param leaseTime 上鎖后自動釋放鎖時間
     * @return boolean
     */
    public boolean tryLock(String lockKey, long waitTime, long leaseTime) {
        RLock lock = redissonClient.getLock(lockKey);
        try {
            return lock.tryLock(waitTime, leaseTime, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error("RedissonUtils - tryLock異常", e);
        }

        return false;
    }

    /**
     * 嘗試獲取鎖
     *
     * @param lockKey   key
     * @param waitTime  最多等待時間
     * @param leaseTime 上鎖后自動釋放鎖時間
     * @param unit      時間單位
     * @return boolean
     */
    public boolean tryLock(String lockKey, long waitTime, long leaseTime, TimeUnit unit) {
        RLock lock = redissonClient.getLock(lockKey);
        try {
            return lock.tryLock(waitTime, leaseTime, unit);
        } catch (InterruptedException e) {
            log.error("RedissonUtils - tryLock異常", e);
        }

        return false;
    }

    /**
     * 釋放鎖
     *
     * @param lockKey key
     */
    public void unlock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.unlock();
    }

    /**
     * 是否存在鎖
     *
     * @param lockKey key
     * @return
     */
    public boolean isLocked(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        return lock.isLocked();
    }
}

2. lock和tryLock的區(qū)別

1.返回值

lock 是 void；
tryLock 是 boolean。

2.時機

lock 一直等鎖釋放；
tryLock 獲取到鎖返回true，獲取不到鎖并直接返回false。

lock拿不到鎖會一直等待。tryLock是去嘗試，拿不到就返回false，拿到返回true。

tryLock是可以被打斷的，被中斷的，lock是不可以。

3. 源碼分析

3.1 lock

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
		// 獲取當(dāng)前線程 ID
        long threadId = Thread.currentThread().getId();
		// 獲取鎖，正常獲取鎖則ttl為null，競爭鎖時返回鎖的過期時間
        Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return;
        }

		// 訂閱鎖釋放事件
		// 如果當(dāng)前線程通過 Redis 的 channel 訂閱鎖的釋放事件獲取得知已經(jīng)被釋放，則會發(fā)消息通知待等待的線程進行競爭
        RFuture<RedissonLockEntry> future = subscribe(threadId);
        if (interruptibly) {
            commandExecutor.syncSubscriptionInterrupted(future);
        } else {
            commandExecutor.syncSubscription(future);
        }

        try {
            while (true) {
                // 循環(huán)重試獲取鎖，直至重新獲取鎖成功才跳出循環(huán)
                // 此種做法阻塞進程，一直處于等待鎖手動釋放或者超時才繼續(xù)線程
                ttl = tryAcquire(-1, leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    break;
                }

                // waiting for message
                if (ttl >= 0) {
                    try {
                        future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        if (interruptibly) {
                            throw e;
                        }
                        future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    }
                } else {
                    if (interruptibly) {
                        future.getNow().getLatch().acquire();
                    } else {
                        future.getNow().getLatch().acquireUninterruptibly();
                    }
                }
            }
        } finally {
        	// 最后釋放訂閱事件
            unsubscribe(future, threadId);
        }
//        get(lockAsync(leaseTime, unit));
    }

    <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                        "redis.call('hincrby', 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.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
    }

此段腳本為一段lua腳本：

KEY[1]: 為你加鎖的lock值

ARGV[2]: 為線程id

ARGV[1]: 為設(shè)置的過期時間

第一個if：

判斷是否存在設(shè)置lock的key是否存在，不存在則利用redis的hash結(jié)構(gòu)設(shè)置一個hash，值為1，并設(shè)置過期時間，后續(xù)返回鎖。

第二個if：

判斷是否存在設(shè)置lock的key是否存在，存在此線程的hash，則為這個鎖的重入次數(shù)加1（將hash值+1），并重新設(shè)置過期時間，后續(xù)返回鎖。

最后返回：

這個最后返回不是說最后結(jié)果返回，是代表以上兩個if都沒有進入，則代表處于競爭鎖的情況，后續(xù)返回競爭鎖的過期時間。

3.2 tryLock

tryLock具有返回值，true或者false，表示是否成功獲取鎖。

tryLock前期獲取鎖邏輯基本與lock一致，主要是后續(xù)獲取鎖失敗的處理邏輯與lock不一致。

    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();
        Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return true;
        }
        
        // 獲取鎖失敗后，中途tryLock會一直判斷中間操作耗時是否已經(jīng)消耗鎖的過期時間，如果消耗完則返回false
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
        
        current = System.currentTimeMillis();
        // 訂閱鎖釋放事件
        // 如果當(dāng)前線程通過 Redis 的 channel 訂閱鎖的釋放事件獲取得知已經(jīng)被釋放,則會發(fā)消息通知待等待的線程進行競爭.
        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        // 將訂閱阻塞，阻塞時間設(shè)置為我們調(diào)用tryLock設(shè)置的最大等待時間，超過時間則返回false
        if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.onComplete((res, e) -> {
                    if (e == null) {
                        unsubscribe(subscribeFuture, threadId);
                    }
                });
            }
            acquireFailed(waitTime, unit, threadId);
            return false;
        }

        try {
            time -= System.currentTimeMillis() - current;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }
        	
        	// 循環(huán)獲取鎖，但由于上面有最大等待時間限制，基本會在上面返回false
            while (true) {
                long currentTime = System.currentTimeMillis();
                ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    return true;
                }

                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(waitTime, unit, threadId);
                    return false;
                }

                // waiting for message
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }

                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(waitTime, unit, threadId);
                    return false;
                }
            }
        } finally {
            unsubscribe(subscribeFuture, threadId);
        }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
    }

應(yīng)盡量使用tryLock，且攜帶參數(shù)，因為可設(shè)置最大等待時間以及可及時獲取加鎖返回值，后續(xù)可做一些其他加鎖失敗的業(yè)務(wù)