Redis結(jié)合Lua腳本實(shí)現(xiàn)分布式鎖詳解

更新時(shí)間：2024年02月28日 10:11:17 作者：coffee_baby

Lua?是一種輕量小巧的腳本語言,用標(biāo)準(zhǔn)C語言編寫并以源代碼形式開放,?本文主要為大家介紹了Redis如何結(jié)合Lua腳本實(shí)現(xiàn)分布式鎖,需要的可以參考下

先講一下為什么使用分布式鎖

在傳統(tǒng)的單體應(yīng)用中，我們可以使用Java并發(fā)處理相關(guān)的API（如ReentrantLock或synchronized）來實(shí)現(xiàn)對共享資源的互斥控制，確保在高并發(fā)情況下同一時(shí)間只有一個線程能夠執(zhí)行特定方法。然而，隨著業(yè)務(wù)的發(fā)展，單體應(yīng)用逐漸演化為分布式系統(tǒng)，多線程、多進(jìn)程分布在不同機(jī)器上，這導(dǎo)致了原有的單機(jī)部署下的并發(fā)控制策略失效。為了解決這一問題，我們需要引入一種跨JVM的互斥機(jī)制來管理共享資源的訪問，這就是分布式鎖所要解決的核心問題。

Lua介紹

Lua 是一種輕量小巧的腳本語言，用標(biāo)準(zhǔn)C語言編寫并以源代碼形式開放，其設(shè)計(jì)目的是為了嵌入應(yīng)用程序中，從而為應(yīng)用程序提供靈活的擴(kuò)展和定制功能。

為什么要用Lua呢

Redis采用單線程架構(gòu)，可以保證單個命令的原子性，但是無法保證一組命令在高并發(fā)場景下的原子性。

在以下場景中：

當(dāng) 事務(wù)1執(zhí)行刪除操作時(shí)，查詢到的鎖值確實(shí)相等。
在事務(wù)1執(zhí)行刪除操作之前，鎖的過期時(shí)間剛好到達(dá)，導(dǎo)致 Redis 自動釋放了該鎖。
事務(wù)2獲取了這個已被釋放的鎖。
當(dāng) 事務(wù)1執(zhí)行刪除操作時(shí)，會意外地刪除掉事務(wù)2持有的鎖。

上面的刪除情況也無法保證原子性，只能通過lua腳本實(shí)現(xiàn)

如果redis客戶端通過lua腳本把3個命令一次性發(fā)送給redis服務(wù)器，那么這三個指令就不會被其他客戶端指令打斷。Redis 也保證腳本會以原子性(atomic)的方式執(zhí)行：當(dāng)某個腳本正在運(yùn)行的時(shí)候，不會有其他腳本或 Redis 命令被執(zhí)行。

Lua腳本命令

在Redis中需要通過eval命令執(zhí)行l(wèi)ua腳本

EVAL script numkeys key [key ...] arg [arg ...]

script：lua腳本字符串，這段Lua腳本不需要（也不應(yīng)該）定義函數(shù)。
numkeys：lua腳本中KEYS數(shù)組的大小
key [key ...]：KEYS數(shù)組中的元素
arg [arg ...]：ARGV數(shù)組中的元素

案列1：動態(tài)傳參

EVAL "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 5 8 10 30 40 50 60 70 
# 輸出：8 10 60 70

EVAL "if KEYS[1] > ARGV[1] then return 1 else return 0 end" 1 10 20
# 輸出：0

EVAL "if KEYS[1] > ARGV[1] then return 1 else return 0 end" 1 20 10
# 輸出：1

案列2：執(zhí)行redis類庫方法

EVAL "return redis.call('set', KEYS[1], ARGV[1])" 1 bbb 20

可重入性

可重入性是指一個線程在持有鎖的情況下，可以多次獲取同一個鎖而不會發(fā)生死鎖或阻塞的特性。在可重入鎖中，線程可以重復(fù)獲取已經(jīng)持有的鎖，每次獲取都會增加一個計(jì)數(shù)器，直到計(jì)數(shù)器歸零時(shí)才會真正釋放鎖。

下面是一個示例代碼來說明可重入性：

public synchronized void a() {
 b();
}
public synchronized void b() {
 // pass
}

假設(shè)線程X在方法a中獲取了鎖后，繼續(xù)執(zhí)行方法b。如果這是一個不可重入的鎖，線程X在執(zhí)行b方法時(shí)將會被阻塞，因?yàn)樗呀?jīng)持有了該鎖并且無法再次獲取。這種情況下，線程X必須等待自己釋放鎖后才能再次爭搶該鎖。

而對于可重入性的情況，當(dāng)線程X持有了該鎖后，在遇到加鎖方法時(shí)會直接將加鎖次數(shù)加1，并繼續(xù)執(zhí)行方法邏輯。當(dāng)退出加鎖方法時(shí)，加鎖次數(shù)再減1。只有當(dāng)加鎖次數(shù)歸零時(shí)，該線程才會真正釋放該鎖。

因此，可重入性的最大特點(diǎn)就是計(jì)數(shù)器的存在，用于統(tǒng)計(jì)加鎖的次數(shù)。在分布式環(huán)境中實(shí)現(xiàn)可重入分布式鎖時(shí)也需要考慮如何正確統(tǒng)計(jì)和管理加鎖次數(shù)。

加鎖腳本

Redis 提供了 Hash （哈希表）這種可以存儲鍵值對數(shù)據(jù)結(jié)構(gòu)。所以我們可以使用 Redis Hash 存儲的鎖的重入次數(shù)，然后利用 lua 腳本判斷邏輯。

if (redis.call('exists', KEYS[1]) == 0 or redis.call('hexists', KEYS[1], ARGV[1]) == 1) 
then
    redis.call('hincrby', KEYS[1], ARGV[1], 1);
    redis.call('expire', KEYS[1], ARGV[2]);
    return 1;
else
	return 0;
end

假設(shè)值為：KEYS:[lock], ARGV[uuid, expire]

如果鎖不存在或者這是自己的鎖，就通過hincrby（不存在就新增并加1，存在就加1）獲取鎖或者鎖次數(shù)加1。

解鎖腳本

-- 判斷 hash set 可重入 key 的值是否等于 0
-- 如果為 nil 代表 自己的鎖已不存在，在嘗試解其他線程的鎖，解鎖失敗
-- 如果為 0 代表 可重入次數(shù)被減 1
-- 如果為 1 代表 該可重入 key 解鎖成功
if(redis.call('hexists', KEYS[1], ARGV[1]) == 0) then 
    return nil; 
elseif(redis.call('hincrby', KEYS[1], ARGV[1], -1) > 0) then 
    return 0; 
else 
    redis.call('del', KEYS[1]); 
    return 1; 
end;

如果鎖不存在直接返回null，如果鎖存在就對數(shù)量進(jìn)行減一，如果減到等于0 就直接刪除此鎖

自動續(xù)期

有可能代碼沒執(zhí)行完畢，鎖就到期了?；谏厦孢@種情況需要對鎖進(jìn)行續(xù)期。使用定時(shí)器加lua腳本進(jìn)行對鎖續(xù)期

if(redis.call('hexists', KEYS[1], ARGV[1]) == 1) then 
    redis.call('expire', KEYS[1], ARGV[2]); 
    return 1; 
else 
    return 0; 
end

Java代碼實(shí)現(xiàn)

考慮到分布式鎖可能使用多種方式實(shí)現(xiàn)，比如Redis、mysql、zookeeper,所以暫時(shí)做成一個工廠類，按需使用。

以下是完整代碼：

public class DistributedRedisLock implements Lock {

    private StringRedisTemplate redisTemplate;

    private String lockName;

    private String uuid;

    private long expire = 30;

    public DistributedRedisLock(StringRedisTemplate redisTemplate, String lockName, String uuid) {
        this.redisTemplate = redisTemplate;
        this.lockName = lockName;
        this.uuid = uuid + ":" + Thread.currentThread().getId();
    }

    @Override
    public void lock() {
        this.tryLock();
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public boolean tryLock() {
        try {
            return this.tryLock(-1L, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 加鎖方法
     * @param time
     * @param unit
     * @return
     * @throws InterruptedException
     */
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        if (time != -1){
            this.expire = unit.toSeconds(time);
        }
        String script = "if redis.call('exists', KEYS[1]) == 0 or redis.call('hexists', KEYS[1], ARGV[1]) == 1 " +
                "then " +
                "   redis.call('hincrby', KEYS[1], ARGV[1], 1) " +
                "   redis.call('expire', KEYS[1], ARGV[2]) " +
                "   return 1 " +
                "else " +
                "   return 0 " +
                "end";
        while (!this.redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), uuid, String.valueOf(expire))){
            Thread.sleep(50);
        }
        // 加鎖成功，返回之前，開啟定時(shí)器自動續(xù)期
        this.renewExpire();
        return true;
    }

    /**
     * 解鎖方法
     */
    @Override
    public void unlock() {
        String script = "if redis.call('hexists', KEYS[1], ARGV[1]) == 0 " +
                "then " +
                "   return nil " +
                "elseif redis.call('hincrby', KEYS[1], ARGV[1], -1) == 0 " +
                "then " +
                "   return redis.call('del', KEYS[1]) " +
                "else " +
                "   return 0 " +
                "end";
        Long flag = this.redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockName), uuid);
        if (flag == null){
            throw new IllegalMonitorStateException("this lock doesn't belong to you!");
        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    private void renewExpire(){
        String script = "if redis.call('hexists', KEYS[1], ARGV[1]) == 1 " +
                "then " +
                "   return redis.call('expire', KEYS[1], ARGV[2]) " +
                "else " +
                "   return 0 " +
                "end";
        new Timer().schedule(new TimerTask() {
            @Override
            public void run() {
                if (redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), uuid, String.valueOf(expire))) {
                    renewExpire();
                }
            }
        }, this.expire * 1000 / 3);
    }
}

DistributedLockClient

@Component
public class DistributedLockClient {
    @Autowired
    private StringRedisTemplate redisTemplate;

    private String uuid;

    public DistributedLockClient() {
        this.uuid = UUID.randomUUID().toString();
    }

    public DistributedRedisLock getRedisLock(String lockName){
        return new DistributedRedisLock(redisTemplate, lockName, uuid);
    }
}

使用及測試：

在業(yè)務(wù)代碼中使用：

public void deduct() {
    DistributedRedisLock redisLock = this.distributedLockClient.getRedisLock("lock");
    redisLock.lock();

    try {
        // 1. 查詢庫存信息
        String stock = redisTemplate.opsForValue().get("stock").toString();

        // 2. 判斷庫存是否充足
        if (stock != null && stock.length() != 0) {
            Integer st = Integer.valueOf(stock);
            if (st > 0) {
                // 3.扣減庫存
                redisTemplate.opsForValue().set("stock", String.valueOf(--st));
            }
        }
    } finally {
        redisLock.unlock();
    }
}

測試可重入性：