1. Redis が実裝する分散ロックの原理

分散ロックが必要な理由

分散ロックについて話す前に、なぜ分散ロックが必要なのかを説明する必要があります。 分散ロック。

分散ロック、スタンドアロン ロックと比較します。マルチスレッド プログラムを作成する場合、共有変數(shù)を同時に操作することによって引き起こされるデータの問題を回避します。通常はロックを使用して相互に排除し、確実なロックを実現(xiàn)します。共有変數(shù).プロパティの正當性、その使用範囲は同じプロセス內です。共有リソースを同時に操作する必要があるプロセスが複數(shù)ある場合、どうすれば相互排他的になるのでしょうか?今日のビジネス アプリケーションは通常マイクロサービス アーキテクチャであり、これは 1 つのアプリケーションが複數(shù)のプロセスをデプロイすることも意味します。複數(shù)のプロセスが MySQL の同じレコード行を変更する必要がある場合、順序の亂れた操作によって引き起こされるダーティ データを避けるために、分散が必要です。今回導入するスタイルはロックされています。

#分散ロックを実裝したい場合は、外部システムを使用する必要があります。すべてのプロセスはこのシステムにアクセスしてロックを申請します。この外部システムは相互に排他的である必要があります。つまり、2 つのリクエストが同時に到著した場合、システムは 1 つのプロセスのみを正常にロックし、もう 1 つのプロセスは失敗します。この外部システムにはデータベース、Redis、Zookeeper のいずれかを使用できますが、パフォーマンスを追求するために、通常は Redis または Zookeeper を使用することを選択します。

Redis は共有ストレージ システムとして使用でき、複數(shù)のクライアントがアクセスを共有できるため、分散ロックの保存に使用できます。さらに、Redis は高い読み取りおよび書き込みパフォーマンスを備えており、同時実行性の高いロック操作シナリオを処理できます。この記事の焦點は、Redis を使用して分散ロックを実裝する方法を紹介し、実裝プロセス中に発生する可能性のある問題について説明することです。

分散ロックの実裝方法

分散ロックの実裝における共有ストレージ システムとして、Redis はキーと値のペアを使用してロック変數(shù)を保存し、受信して処理できます。さまざまなクライアントから送信されるロックとロックの解放の操作リクエスト。では、キーと値のペアのキーと値はどのように決定されるのでしょうか?ロック変數(shù)に変數(shù)名を付け、この変數(shù)名をキーと値のペアのキーとして使用する必要があり、ロック変數(shù)の値はキーと値のペアの値になります。このようにして、Redis はロック変數(shù)とクライアントは、Redis コマンド操作を通じてロック操作を実裝できます。

分散ロックを実裝するには、Redis に相互排他機能が必要です。 SETNX コマンドを使用できます。これは SET IF NOT EXIST を意味します。つまり、キーが存在しない場合はその値が設定され、それ以外の場合は何も行われません。分散ロックは、2 つのクライアント プロセスにコマンドを相互に排他的に実行させることで実裝されます。

以下は、キーと値のペアを使用してロック変數(shù)を保存する Redis の操作プロセスと、同時にロックを要求する 2 つのクライアントを示しています。

#ロック操作が完了すると、ロックに成功したクライアントは共有リソースを操作できるようになります。たとえば、MySQL の特定のデータ行を変更できます。操作が完了したら、後発者に共有リソースを操作する機會を與えるために、ロックを時間內に解放する必要があります。ロックを解除するにはどうすればよいですか?このキーを削除するには、DEL コマンドを使用するだけです。ロジックは非常にシンプルで、全體の処理を擬似コードで記述すると以下のようになります。

// 加鎖
SETNX lock_key 1
// 業(yè)務邏輯
DO THINGS
// 釋放鎖
DEL lock_key

ただし、上記の実裝には大きな問題があり、クライアント 1 がロックを取得したときに、次のような狀況が発生するとデッドロックが発生します。

プログラムはビジネス ロジック例外を処理し、時間內にロックを解放できません。プロセスがハングし、ロックを解放する機會がありません。

上記の狀況により、クライアントがlock を使用すると、ロックを永久に占有し、他のクライアントは決してそのロックを取得できなくなります。

デッドロックを回避する方法

上記のデッドロックの問題を解決するために考えられる最も簡単な解決策は、ロックを申請するときにそのロックにロックを設定することです。有効期限は、共有リソースの操作時間が 10 秒を超えないと仮定し、ロックする場合は、このキーの有効期限を 10 秒に設定するだけです。

しかし、上記の操作にはまだ問題があります。ロックして有効期限を設定するコマンドは 2 つあります。最初のコマンドだけが実行され、2 番目のコマンドは実行に失敗する可能性があります、例:

1. SETNX は正常に実行されましたが、ネットワークの問題により EXPIRE は失敗しました。
2. SETNX は正常に実行されましたが、Redis が異常にクラッシュし、EXPIRE にはチャンスがありませんでした
3. SETNX は正常に実行され、顧客は端末が異常にクラッシュし、EXPIRE を実行する機會がありませんでした

つまり、これら 2 つのコマンドがアトミックであることが保証できない場合は、操作を実行すると、有効期限の設定が失敗し、デッドロックの問題が依然として発生する可能性がある潛在的なリスクがあります。幸いなことに、Redis 2.6.12 以降、Redis は SET コマンドのパラメーターを拡張しました。SET と同時に EXPIRE 時間を指定できます。この操作はアトミックです。たとえば、次のコマンドはロックの有効期限を 10 秒に設定します。

SET lock_key 1 EX 10 NX

これまでのところ、デッドロックの問題は解決されましたが、他にも問題がまだ殘っています。次のシナリオを想像してください:

クライアント 1 は正常にロックされ、共有リソースの操作を開始します

2. 客戶端1操作共享資源耗時太久，超過了鎖的過期時間，鎖失效（鎖被自動釋放）

3. 客戶端2加鎖成功，開始操作共享資源

4. 客戶端1操作共享資源完成，在finally塊中手動釋放鎖，但此時它釋放的是客戶端2的鎖。

這里存在兩個嚴重的問題：

• 鎖過期

• 釋放了別人的鎖

第1個問題是評估操作共享資源的時間不準確導致的，如果只是一味增大過期時間，只能緩解問題降低出現(xiàn)問題的概率，依舊無法徹底解決問題。原因在于客戶端在拿到鎖之后，在操作共享資源時，遇到的場景是很復雜的，既然是預估的時間，也只能是大致的計算，不可能覆蓋所有導致耗時變長的場景。

第二個問題在于解鎖操作是不夠嚴謹?shù)?，因為它是一種不加區(qū)分地釋放鎖的操作，沒有對鎖的所有權進行檢查。如何解決呢？

鎖被別人給釋放了

解決辦法是，客戶端在加鎖時，設置一個只有自己知道的唯一標識進去，例如可以是自己的線程ID，如果是redis實現(xiàn)，就是SET key unique_value EX 10 NX。之后在釋放鎖時，要先判斷這把鎖是否歸自己持有，只有是自己的才能釋放它。

//釋放鎖 比較unique_value是否相等，避免誤釋放
if redis.get("key") == unique_value then
    return redis.del("key")

這里釋放鎖使用的是GET + DEL兩條命令，這時又會遇到原子性問題了。

1. 客戶端1執(zhí)行GET，判斷鎖是自己的

2. 客戶端2執(zhí)行了SET命令，強制獲取到鎖（雖然發(fā)生概念很低，但要嚴謹考慮鎖的安全性）

3. 客戶端1執(zhí)行DEL，卻釋放了客戶端2的鎖

由此可見，以上GET + DEL兩個命令還是必須原子的執(zhí)行才行。怎樣原子執(zhí)行兩條命令呢？答案是Lua腳本，可以把以上邏輯寫成Lua腳本，讓Redis執(zhí)行。因為Redis處理每個請求是單線程執(zhí)行的，在執(zhí)行一個Lua腳本時其它請求必須等待，直到這個Lua腳本處理完成，這樣一來GET+DEL之間就不會有其他命令執(zhí)行了。

以下是使用Lua腳本（unlock.script）實現(xiàn)的釋放鎖操作的偽代碼，其中，KEYS[1]表示lock_key，ARGV[1]是當前客戶端的唯一標識，這兩個值都是我們在執(zhí)行 Lua腳本時作為參數(shù)傳入的。

//Lua腳本語言，釋放鎖 比較unique_value是否相等，避免誤釋放
if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end

最后我們執(zhí)行以下命令，即可

redis-cli  --eval  unlock.script lock_key , unique_value

這樣一路優(yōu)先下來，整個加鎖、解鎖流程就更嚴謹了，先小結一下，基于Redis實現(xiàn)的分布式鎖，一個嚴謹?shù)牧鞒倘缦拢?/p>

1. 加鎖時要設置過期時間SET lock_key unique_value EX expire_time NX

2. 操作共享資源

3. 釋放鎖：Lua腳本，先GET判斷鎖是否歸屬自己，再DEL釋放鎖

有了這個嚴謹?shù)逆i模型，我們還需要重新思考之前的那個問題，鎖的過期時間不好評估怎么辦。

如何確定鎖的過期時間

前面提到過，過期時間如果評估得不好，這個鎖就會有提前過期的風險，一種妥協(xié)的解決方案是，盡量冗余過期時間，降低鎖提前過期的概率，但這個方案并不能完美解決問題。是否可以設置這樣的方案，加鎖時，先設置一個預估的過期時間，然后開啟一個守護線程，定時去檢測這個鎖的失效時間，如果鎖快要過期了，操作共享資源還未完成，那么就自動對鎖進行續(xù)期，重新設置過期時間。

Redisson是一個已封裝好這些工作的庫，可以說是一種非常優(yōu)秀的解決方案。Redisson是一個Java語言實現(xiàn)的Redis SDK客戶端，在使用分布式鎖時，它就采用了自動續(xù)期的方案來避免鎖過期，這個守護線程我們一般叫它看門狗線程。這個SDK提供的API非常友好，它可以像操作本地鎖一樣操作分布式鎖?？蛻舳艘坏┘渔i成功，就會啟動一個watch dog看門狗線程，它是一個后臺線程，會每隔一段時間（這段時間的長度與設置的鎖的過期時間有關）檢查一下，如果檢查時客戶端還持有鎖key（也就是說還在操作共享資源），那么就會延長鎖key的生存時間。

那如果客戶端在加鎖成功后就宕機了呢？宕機了那么看門狗任務就不存在了，也就無法為鎖續(xù)期了，鎖到期自動失效。

Redis的部署方式對鎖的影響

上面討論的情況，都是鎖在單個Redis 實例中可能產(chǎn)生的問題，并沒有涉及到Redis的部署架構細節(jié)。

Redis發(fā)展到現(xiàn)在，幾種常見的部署架構有：

• #シングル モード;

• マスター/スレーブ モード;

• センチネル モード;

#クラスター モード;
Redis を使用する場合、

通常、マスター/スレーブ クラスター センチネル モードでデプロイします。センチネルの役割は、 redis ノードの動作ステータス。通常のマスター/スレーブ モードでは、マスターがクラッシュした場合、手動でスレーブをマスターに切り替える必要がありますが、マスターとスレーブのセンチネルの組み合わせを使用する利點は、マスターが異常にクラッシュした場合にセンチネルが自動フェイルオーバーを実裝し、スレーブを新しいマスターに昇格させます?？捎眯?p>はサービスを提供し続けることで保証されます。それでは、マスターとスレーブの切り替えが発生したとき、分散ロックは依然として安全なのでしょうか?

#次のシナリオを想像してください:

クライアント 1 がマスター上で SET コマンドを実行し、ロックが成功します

1. 現(xiàn)時點では、マスターは異常ダウンしており、SET コマンドはまだスレーブに同期されていません (マスター/スレーブ レプリケーションは非同期です)

2. Sentinel はスレーブを新しいマスターに昇格させますが、新しいマスターでロックが失われたため、クライアント 2 は正常にロックされますが、分散ロックは依然として影響を受ける可能性があります。 Redis がセンチネルを通じて高可用性を確保したとしても、何らかの理由でマスター ノードがマスターとスレーブを切り替えると、ロックは失われます。

3. クラスターモード Redlock は信頼性の高い分散ロックを実裝します

Redis インスタンスの障害によるロック失敗の問題を回避するために、Redis 開発者 Antirez は分散フォーミュラ ロック アルゴリズム Redlock を提案しました。 。 Redlock アルゴリズムの基本的な考え方は、クライアントと複數(shù)の獨立した Redis インスタンスが順番にロックを要求できるようにすることです。クライアントが半分以上のインスタンスでロック操作を正常に完了できた場合、クライアントは成功したと見なされます。分散ロックが取得されると、ロックは失敗します

。このように、単一の Redis インスタンスに障害が発生した場合でも、ロック変數(shù)は他のインスタンスにも保存されるため、クライアントは引き続きロック操作を正常に実行でき、ロック変數(shù)が失われることはありません。

Redlock アルゴリズムの実行ステップを詳しく見てみましょう。 Redlock アルゴリズムを実裝するには、Redis がセンチネル ノードを使用しないクラスター デプロイメント モードを採用し、N 個の獨立した Redis インスタンス (公式には少なくとも 5 つのインスタンスが推奨されます) を必要とします。次に、ロック操作は 3 ステップで完了します。

最初のステップは、クライアントが現(xiàn)在時刻を取得することです。

2 番目のステップは、クライアントが N 個の Redis インスタンスに対してロック操作を順番に実行することです。

ここでのロック操作は、単一インスタンスで実行されるロック操作と同じです。NX、EX/PX オプション、およびクライアントの一意の識別子を指定して SET コマンドを使用します。もちろん、Redis インスタンスに障害が発生した場合でも Redlock アルゴリズムが確実に実行できるようにするには、ロック操作のタイムアウトを設定する必要があります。クライアントがタイムアウトになるまで Redis インスタンスでのロックの要求に失敗した場合、この時點でクライアントは次の Redis インスタンスでのロックを要求し続けます。一般に、ロック操作のタイムアウトは、ロックの有効時間のごく一部 (通常は約數(shù)十ミリ秒) に設定する必要があります。

3 番目のステップでは、クライアントがすべての Redis インスタンスでロック操作を完了したら、クライアントはロック プロセス全體に費やした合計時間を計算する必要があります。

クライアントは、2 つの條件が満たされた場合にのみ、ロックが成功したとみなすことができます。條件 1 は、クライアントが半分以上 (N/2 1 以上) からロックを正常に取得したことです。 ) ; 2 番目の條件は、クライアントがロックを取得するのに費やした合計時間が、ロックの有効時間を超えないことです。

ほとんどのインスタンスが正常にロックされた場合にのみ、操作が成功したとみなされるのはなぜですか?実際、複數(shù)の Redis インスタンスを一緒に使用して分散システムを形成します。分散システムには異常なノードが常に存在するため、分散システムについて語るときは、システム全體の正常な動作に影響を與えずに、異常なノードがどれだけ存在するかを考慮する必要があります。これは分散システムにおけるフォールト トレランスの問題であり、この問題の結論は、障害のあるノードだけが存在する場合でも、ほとんどのノードが正常である限り、システム全體は依然として正しいサービスを提供できるということです。

これら 2 つの條件を満たした後、

ロックの有効時間を再計算する必要があります。計算の結果は、ロックの初期有効時間からクライアントがロックを取得するのに費やした合計時間を引いたものになります。ロック。ロックの有効期限が遅すぎて共有データ操作を完了できない場合は、共有リソース操作が完了する前にロックが期限切れになる狀況を避けるためにロックを解放できます。

當然，如果客戶端在和所有實例執(zhí)行完加鎖操作后，沒能同時滿足這兩個條件，那么，客戶端就要向所有Redis節(jié)點發(fā)起釋放鎖的操作。為什么釋放鎖，要操作所有的節(jié)點呢，不能只操作那些加鎖成功的節(jié)點嗎？因為在某一個Redis節(jié)點加鎖時，可能因為網(wǎng)絡原因導致加鎖失敗，例如一個客戶端在一個Redis實例上加鎖成功，但在讀取響應結果時由于網(wǎng)絡問題導致讀取失敗，那這把鎖其實已經(jīng)在Redis上加鎖成功了。所以釋放鎖時，不管之前有沒有加鎖成功，需要釋放所有節(jié)點上的鎖以保證清理節(jié)點上的殘留的鎖。

在Redlock算法中，釋放鎖的操作和在單實例上釋放鎖的操作一樣，只要執(zhí)行釋放鎖的 Lua腳本就可以了。如果N個Redis實例中超過一半的實例正常工作，就能確保分布式鎖正常運作。為了提高分布式鎖的可靠性，您可以在實際業(yè)務應用中使用Redlock算法。

二、代碼實現(xiàn)Redis分布式鎖

1.SpringBoot整合redis用到最多的當然屬于我們的老朋友RedisTemplate，pom依賴如下：

<!-- springboot整合redis -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

2.Redis配置類:

package com.example.redisdemo.config;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

/**
 * @description: Redis配置類
 * @author Keson
 * @date 21:20 2022/11/14
 * @Param
 * @return
 * @version 1.0
 */
@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory) {
        // 設置序列化
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<Object>(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // 配置redisTemplate
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
        redisTemplate.setConnectionFactory(lettuceConnectionFactory);
        RedisSerializer<?> stringSerializer = new StringRedisSerializer();
        redisTemplate.setKeySerializer(stringSerializer);// key序列化
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);// value序列化
        redisTemplate.setHashKeySerializer(stringSerializer);// Hash key序列化
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);// Hash value序列化
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }
}

3.Service層面

package com.example.redisdemo.service;

import com.example.redisdemo.entity.CustomerBalance;
import java.util.concurrent.Callable;

/**
 * @author Keson
 * @version 1.0
 * @description: TODO
 * @date 2022/11/14 15:12
 */
public interface RedisService {

    <T> T callWithLock(CustomerBalance customerBalance, Callable<T> callable) throws Exception;
}

package com.example.redisdemo.service.impl;

import com.example.redisdemo.entity.CustomerBalance;
import com.example.redisdemo.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.stereotype.Service;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

/**
 * @author Keson
 * @version 1.0
 * @description: TODO Redis實現(xiàn)分布式鎖
 * @date 2022/11/14 15:13
 */
@Service
@Slf4j
public class RedisServiceImpl implements RedisService {

    //設置默認過期時間
    private final static int DEFAULT_LOCK_EXPIRY_TIME = 20;
    //自定義lock key前綴
    private final static String LOCK_PREFIX = "LOCK:CUSTOMER_BALANCE";

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public <T> T callWithLock(CustomerBalance customerBalance, Callable<T> callable) throws Exception{
        //自定義lock key
        String lockKey = getLockKey(customerBalance.getCustomerNumber(), customerBalance.getSubAccountNumber(), customerBalance.getCurrencyCode());
        //將UUID當做value，確保唯一性
        String lockReference = UUID.randomUUID().toString();

        try {
            if (!lock(lockKey, lockReference, DEFAULT_LOCK_EXPIRY_TIME, TimeUnit.SECONDS)) {
                throw new Exception("lock加鎖失敗");
            }
            return callable.call();
        } finally {
            unlock(lockKey, lockReference);
        }
    }

    //定義lock key
    String getLockKey(String customerNumber, String subAccountNumber, String currencyCode) {
        return String.format("%s:%s:%s:%s", LOCK_PREFIX, customerNumber, subAccountNumber, currencyCode);
    }

    //redis加鎖
    private boolean lock(String key, String value, long timeout, TimeUnit timeUnit) {
        Boolean locked;
        try {
            //SET_IF_ABSENT --> NX: Only set the key if it does not already exist.
            //SET_IF_PRESENT --> XX: Only set the key if it already exist.
            locked = (Boolean) redisTemplate.execute((RedisCallback<Boolean>) connection ->
                    connection.set(key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8),
                            Expiration.from(timeout, timeUnit), RedisStringCommands.SetOption.SET_IF_ABSENT));
        } catch (Exception e) {
            log.error("Lock failed for redis key: {}, value: {}", key, value);
            locked = false;
        }
        return locked != null && locked;
    }

    //redis解鎖
    private boolean unlock(String key, String value) {
        try {
            //使用lua腳本保證刪除的原子性，確保解鎖
            String script = "if redis.call(&#39;get&#39;, KEYS[1]) == ARGV[1] " +
                            "then return redis.call(&#39;del&#39;, KEYS[1]) " +
                            "else return 0 end";
            Boolean unlockState = (Boolean) redisTemplate.execute((RedisCallback<Boolean>) connection ->
                    connection.eval(script.getBytes(), ReturnType.BOOLEAN, 1,
                            key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8)));
            return unlockState == null || !unlockState;
        } catch (Exception e) {
            log.error("unLock failed for redis key: {}, value: {}", key, value);
            return false;
        }
    }
}

4.業(yè)務調用實現(xiàn)分布式鎖示例：

    @Override
    public int updateById(CustomerBalance customerBalance) throws Exception {
        return redisService.callWithLock(customerBalance, ()-> customerBalanceMapper.updateById(customerBalance));
    }

以上がRedis を使用して SpringBoot に分散ロックを実裝する方法の詳細內容です。詳細については、PHP 中國語 Web サイトの他の関連記事を參照してください。