2.1. 基本實現(xiàn)

借助于redis中的命令setnx(key, value)，key不存在就新增，存在就什么都不做。同時有多個客戶端發(fā)送setnx命令，只有一個客戶端可以成功，返回1（true）；其他的客戶端返回0（false）。

1. 多個客戶端同時獲取鎖（setnx）
2. 獲取成功，執(zhí)行業(yè)務邏輯，執(zhí)行完成釋放鎖（del）
3. 其他客戶端等待重試

改造StockService方法：

 /**
     * redis setnx實現(xiàn)分布式鎖，最基本的哪一種 ！??！
     */
    public void deduct() {
        // 加鎖setnx
        Boolean lock = this.redisTemplate.opsForValue().setIfAbsent("lock", "111");
        if (!lock) {
            // 沒有獲取到鎖，進行重試！！
            try {
                Thread.sleep(50);
                this.deduct();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } else {
            try {
                // 1. 查詢庫存信息
                String stockStr = redisTemplate.opsForValue().get("stock:" + "1001");
                // 2. 判斷庫存是否充足
                if (stockStr != null && stockStr.length() != 0) {
                    Long stock = Long.parseLong(stockStr);
                    if (stock > 0) {
                        redisTemplate.opsForValue().set("stock:" + "1001", String.valueOf(stock - 1));
                    }
                }
            } finally {
                // 解鎖
                this.redisTemplate.delete("lock");
            }
        }
    }

使用 jmeter 進行壓測

查看庫存數(shù)量

上述代碼優(yōu)化，不斷重試的過程中一直進行遞歸，最終導致棧的溢出。

解決

    /**
     *  while循環(huán)代替遞歸，解決不斷重試可能導致的棧溢出的問題
     */
    public void deduct() {
        // 加鎖setnx
        while (this.redisTemplate.opsForValue().setIfAbsent("lock1", "1")) {
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            try {
                // 1. 查詢庫存信息
                String stockStr = redisTemplate.opsForValue().get("stock:" + "1001");
                // 2. 判斷庫存是否充足
                if (stockStr != null && stockStr.length() != 0) {
                    Long stock = Long.parseLong(stockStr);
                    if (stock > 0) {
                        redisTemplate.opsForValue().set("stock:" + "1001", String.valueOf(stock - 1));
                    }
                }
            } finally {
                // 解鎖
                this.redisTemplate.delete("lock1");
            }
        }
    }

2.2. 防死鎖

問題：setnx剛剛獲取到鎖，當前服務器宕機，導致del釋放鎖無法執(zhí)行，進而導致鎖無法鎖無法釋放（死鎖）
解決：給鎖設置過期時間，自動釋放鎖。

設置過期時間兩種方式：

• 通過expire設置過期時間（缺乏原子性：如果在setnx和expire之間出現(xiàn)異常，鎖也無法釋放）
• 使用set指令設置過期時間：set key value ex 3 nx（既達到setnx的效果，又設置了過期時間）

2.3. 防誤刪

持有鎖的線程在鎖的內(nèi)部出現(xiàn)了阻塞，導致他的鎖自動釋放，這時其他線程，線程2來嘗試獲得鎖，就拿到了這把鎖，然后線程2在持有鎖執(zhí)行過程中，線程1反應過來，繼續(xù)執(zhí)行，而線程1執(zhí)行過程中，走到了刪除鎖邏輯，此時就會把本應該屬于線程2的鎖進行刪除，這就是誤刪別人鎖的情況說明

解決： setnx獲取鎖時，設置一個指定的唯一值（例如：uuid）；釋放前獲取這個值，判斷是否自己的鎖

問題：刪除操作缺乏原子性。
場景：

1. index1執(zhí)行刪除時，查詢到的lock值確實和uuid相等
2. index1執(zhí)行刪除前，lock剛好過期時間已到，被redis自動釋放
3. index2獲取了lock
4. index1執(zhí)行刪除，此時會把index2的lock刪除

解決方案：沒有一個命令可以同時做到判斷 + 刪除，所有只能通過其他方式實現(xiàn)（LUA腳本）

2.4. redis中的lua腳本

2.4.1 redis 并不能保證一組命令的原子性

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

如果redis客戶端通過lua腳本把3個命令一次性發(fā)送給redis服務器，那么這三個指令就不會被其他客戶端指令打斷。Redis 也保證腳本會以原子性的方式執(zhí)行： 當某個腳本正在運行的時候，不會有其他腳本或 Redis 命令被執(zhí)行。 這和使用 MULTI/ EXEC 包圍的事務很類似。

2.4.2 lua介紹

Lua 是一種輕量小巧的腳本語言，用標準C語言編寫并以源代碼形式開放， 其設計目的是為了嵌入應用程序中，從而為應用程序提供靈活的擴展和定制功能。

2.4.3. lua基本語法

a = 5               -- 全局變量
local b = 5         -- 局部變量， redis只支持局部變量
a, b = 10, 2*x      -- 等價于       a=10; b=2*x

流程控制：
if( 布爾表達式 1)
then
   --[ 在布爾表達式 1 為 true 時執(zhí)行該語句塊 --]
elseif( 布爾表達式 2)
then
   --[ 在布爾表達式 2 為 true 時執(zhí)行該語句塊 --]
else 
   --[ 如果以上布爾表達式都不為 true 則執(zhí)行該語句塊 --]
end

2.4.4. redis執(zhí)行l(wèi)ua腳本 - EVAL指令

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

格式：

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

案例1：基本案例

EVAL "return 10" 0

輸出：(integer) 10

EVAL "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 5 10 20 30 40 50 60 70 80 90
# 輸出：10 20 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

傳入了兩個參數(shù)10和20，KEYS的長度是1，所以KEYS中有一個元素10，剩余的一個20就是ARGV數(shù)組的元素。

案例3：執(zhí)行redis類庫方法
redis.call()中的redis是redis中提供的lua腳本類庫，僅在redis環(huán)境中可以使用該類庫

set aaa 10  -- 設置一個aaa值為10
EVAL "return redis.call('get', 'aaa')" 0
# 通過return把call方法返回給redis客戶端，打?。?10"

注意：**腳本里使用的所有鍵都應該由 KEYS 數(shù)組來傳遞。**但并不是強制性的，代價是這樣寫出的腳本不能被 Redis 集群所兼容

2.5. 使用lua保證刪除原子性

刪除LUA腳本：

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

代碼實現(xiàn) StockService：

    /**
     *  解決鎖的誤刪問題
     */
    public void deduct() {
        String uuid = UUID.randomUUID().toString();
        // 加鎖setnx
        while (this.redisTemplate.opsForValue().setIfAbsent("lock1", uuid, 20,TimeUnit.SECONDS)) {
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            try {
                // 1. 查詢庫存信息
                String stockStr = redisTemplate.opsForValue().get("stock:" + "1001");
                // 2. 判斷庫存是否充足
                if (stockStr != null && stockStr.length() != 0) {
                    Long stock = Long.parseLong(stockStr);
                    if (stock > 0) {
                        redisTemplate.opsForValue().set("stock:" + "1001", String.valueOf(stock - 1));
                    }
                }
            } finally {
                // 判斷鎖是不是被當前線程所持有，是的話，則刪除鎖
                /*if(uuid.equals(redisTemplate.opsForValue().get("lock1"))) {
                    // 解鎖
                    this.redisTemplate.delete("lock1");
                }*/

                // 通過lua腳本來釋放鎖
                String script = "if redis.call('get', KEYS[1]) == ARGV[1] " +
                        "then " +
                        "   return redis.call('del', KEYS[1]) " +
                        "else " +
                        "   return 0 " +
                        "end";
                this.redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList("lock"), uuid);
            }
        }
    }

2.6. 可重入鎖

由于上述加鎖命令使用了 SETNX ，一旦鍵存在就無法再設置成功，這就導致后續(xù)同一線程內(nèi)繼續(xù)加鎖，將會加鎖失敗。

用一段 Java 代碼解釋可重入：

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

public synchronized void b() {
    // pass
}

假設 X 線程在 a 方法獲取鎖之后，繼續(xù)執(zhí)行 b 方法，如果此時不可重入，線程就必須等待鎖釋放，再次爭搶鎖。
鎖明明是被 X 線程擁有，卻還需要等待自己釋放鎖，然后再去搶鎖，這看起來就很奇怪，我釋放我自己~

可重入性就可以解決這個尷尬的問題，當線程擁有鎖之后，往后再遇到加鎖方法，直接將加鎖次數(shù)加 1，然后再執(zhí)行方法輯。退出加鎖方法之后，加鎖次數(shù)再減 1，當加鎖次數(shù)為 0 時，鎖才被真正的釋放。

2.6.1. 加鎖腳本

參照 ReentrantLock 中的非公平可重入鎖實現(xiàn)分布式可重入鎖： hash + lua腳本

加鎖：
1、判斷鎖是否存在（exists）, 不存在則直接獲取鎖，hset key filed value
2、如果鎖存在則判斷是否是自己的鎖（hexists）
如果是自己的鎖則重入：hincrby key field increment
如果不是自己的鎖，則重試

// 當鎖不存在 或者 持有鎖的線程是當前線程時
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

2.6.2. 解鎖腳本

解鎖
1、判斷自己的鎖是否存在，不存在則返回nil
2、如果自己的鎖存在，則減1（hincrby -1 ），判斷減1后的值是否為0
為0則del key并且返回1，表示解鎖成功
不為0，則返回0表示解鎖失敗

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;

2.6.3. 代碼實現(xiàn)

由于加解鎖代碼量相對較多，這里可以封裝成一個工具類：

// 生成鎖的工廠
@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);
    }
}

// 鎖具體實現(xiàn)
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;
    }

    @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;
    }

    @Override
    public boolean tryLock(long time, TimeUnit timeUnit) throws InterruptedException {
        if (time != -1){
            this.expire = timeUnit.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), getId(), String.valueOf(expire))) {
            Thread.sleep(50);
        }
        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), getId());
        if (flag == null) {
            throw new IllegalMonitorStateException("this lock doesn't belong to you!");
        }
    }

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


    /**
     * 給線程拼接唯一標識
     *
     * @return
     */
    String getId() {
        return uuid + ":" + Thread.currentThread().getId();
    }
}

2.6.4. 使用及測試

在業(yè)務代碼中進行使用：

    public void deduct() {
        DistributedRedisLock lock = distributedLockClient.getRedisLock("lock");
        try {
            lock.lock();
            // 1. 查詢庫存信息
            String stockStr = redisTemplate.opsForValue().get("stock:" + "1001");
            // 2. 判斷庫存是否充足
            if (stockStr != null && stockStr.length() != 0) {
                Long stock = Long.parseLong(stockStr);
                if (stock > 0) {
                    redisTemplate.opsForValue().set("stock:" + "1001", String.valueOf(stock - 1));
                }
            }
        } finally {
            lock.unlock();
        }
    }

測試可重入性：

  public void deduct() {
        DistributedRedisLock lock = distributedLockClient.getRedisLock("lock");
        try {
            lock.lock();
            // 1. 查詢庫存信息
            String stockStr = redisTemplate.opsForValue().get("stock:" + "1001");
            // 2. 判斷庫存是否充足
            if (stockStr != null && stockStr.length() != 0) {
                Long stock = Long.parseLong(stockStr);
                if (stock > 0) {
                    redisTemplate.opsForValue().set("stock:" + "1001", String.valueOf(stock - 1));
                }
            }
            this.test();
        } finally {
            lock.unlock();
        }
    }

    private void test() {
        DistributedRedisLock lock = distributedLockClient.getRedisLock("lock");
        lock.lock();
        System.out.println("獲取到可重入鎖");
        System.out.println("執(zhí)行其他業(yè)務代碼......");
        lock.unlock();
    }

2.7. 自動續(xù)期

自動續(xù)期： 定時任務 （Time 定時器 ） + Lua腳本

判斷自己的鎖是否存在（hexists），如果存在則重置過期時間

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

DistributedRedisLock 實現(xiàn)

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;
    }

    @Override
    public boolean tryLock(long time, TimeUnit timeUnit) throws InterruptedException {
        if (time != -1) {
            this.expire = timeUnit.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);
        }
        // 加鎖成功返回之前，開啟自動續(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))) {
                    // 當續(xù)期成功了，重新開啟一個定時任務
                    renewExpire();
                }
            }
        }, this.expire * 1000 / 3);
    }
}

測試 ，改造stockService

    public void deduct() {
        DistributedRedisLock lock = distributedLockClient.getRedisLock("lock");
        try {
            lock.lock();
            // 1. 查詢庫存信息
            try {
                Thread.sleep(300000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            String stockStr = redisTemplate.opsForValue().get("stock:" + "1001");
            // 2. 判斷庫存是否充足
            if (stockStr != null && stockStr.length() != 0) {
                Long stock = Long.parseLong(stockStr);
                if (stock > 0) {
                    redisTemplate.opsForValue().set("stock:" + "1001", String.valueOf(stock - 1));
                }
            }
            this.test();
        } finally {
            lock.unlock();
        }
    }

可以看到加鎖 10s后，鎖過期時間重置為30s

【注意】
鎖的自動續(xù)期和鎖的過期釋放是不是沖突的？

鎖的自動續(xù)期和鎖的過期釋放并不沖突，它們是為了解決不同場景下的問題而設計的。

鎖的自動續(xù)期：這個機制主要用于處理持有鎖的客戶端在執(zhí)行任務時可能因為某些原因導致任務執(zhí)行時間超過預設值。通過自動續(xù)期，可以確保在客戶端仍在執(zhí)行任務時，其他客戶端無法獲取到該鎖。這樣就避免了多個客戶端同時操作共享資源造成數(shù)據(jù)混亂或者錯誤。

鎖的過期釋放：這個機制主要用于處理持有鎖的客戶端異常退出（比如崩潰、斷電等）導致無法正常釋放鎖。通過設置一個合理的過期時間，在發(fā)生異常情況后一段時間內(nèi)其他客戶端依然可以獲取到該鎖，并且進行相應操作。

總之，兩者都是為了確保分布式系統(tǒng)中對共享資源訪問互斥性以及容錯性而設計，并不存在沖突關系。文章來源地址http://www.zghlxwxcb.cn/news/detail-617374.html

到了這里，關于2、基于redis實現(xiàn)分布式鎖的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！