SpringBootとデータベーステーブルレコードに基づく分散ロックの実装

[[386855]]

同じプロセス内の異なるスレッドが共有リソースを操作する場合、操作の正確性を確保するために、JUC のツールを使用するなどしてリソースをロックするだけで済みます。 JUC についてよく知らない方は、以下の記事をお読みください。

同期の簡単な紹介
同期最適化
JUC の礎 - 安全でないクラス

ただし、高可用性を実現するために、システムでは常に複数のコピーが別のマシンに分散されているため、同じプロセス内での上記のロックメカニズムは機能しなくなります。マルチコピーシステムによる共有リソースへのアクセスを保証するために、分散ロックを導入しました。

分散ロックの主な実装方法は次のとおりです。

データベースベース。これはさらに、データベースベースのテーブルレコード、悲観的ロック、楽観的ロックに分類されます。
Redisなどのキャッシュベース
Zookeeperに基づく

今日は、最もシンプルな分散ロックソリューションである、データベーステーブルレコードに基づく分散ロックを紹介します。

主な原則は、データベースの一意のインデックスを使用することです（データベースのインデックスについて知らない人は、私の別の記事「MySQLインデックスに関する簡単な説明」を参照してください）。

たとえば、次の表があります。

作成する テーブル`test`.`Untitled` (
  `id` int (11)ではない  NULL AUTO_INCREMENT COMMENT 'シリアル番号の自動増分' ,
  ` name ` varchar (255) NOT   NULL COMMENT 'ロック名' 、
  `survival_time` int (11)ではない  NULL COMMENT '生存時間（ミリ秒）' 、
  `create_time`タイムスタンプ(3) NOT   NULL  デフォルト  CURRENT_TIMESTAMP (3) COMMENT '作成時刻' 、
  `thread_name` varchar (255) NOT   NULL COMMENT 'スレッド名' 、
主要な  KEY (`id`) BTREEの使用、
個性的 インデックス`uk_name`(` name `) BTREE 使用
) ENGINE = InnoDB ROW_FORMAT =動的;

名前フィールドに一意のインデックスが追加されます。同じ名前値を持つ複数の新しい操作の場合、データベースは 1 つの操作のみが成功することだけを保証できます。その他の操作は拒否され、「重複キー」エラーがスローされます。

次に、システム 1 が分散ロックを取得する準備ができたら、name="key" のレコードをデータベースに挿入しようとします。挿入が成功した場合、ロックが正常に取得されたことを意味します。他のシステムが分散ロックを取得する場合は、同じ名前のレコードをデータベースに挿入する必要もあります。もちろん、データベースはエラーを報告し、挿入は失敗します。これは、これらのシステムがロックを取得できなかったことを意味します。システム 1 がロックを解除したい場合、このレコードを削除できます。 thread_name 列を使用すると、スレッドによって作成されたロックのみがアクティブに解放されることを確認できます。

私たちが実現したい分散ロックには、次のような効果があります。

ロックを取得することはブロックであり、取得できない場合はブロックされます。
ロックの有効期間が終了すると、ロックは無効になり、自動的に解除されます。これにより、ダウンタイムのために一部のシステムがロックを積極的に解放できないという問題を回避できます。

一般的なフローチャートは次のとおりです。

次の依存関係が使用されます。

スプリングブート
MyBatisプラス
ロンボク

プロジェクトディレクトリは次のとおりです。

pom ファイルで使用される依存関係は次のとおりです。

 <依存関係>
    <依存関係>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </依存関係> 
 
    <依存関係>
        <groupId>org.projectlombok</groupId>
        <artifactId>ロンボク</artifactId>
        <バージョン>1.18.6</バージョン>
    </依存関係> 
 
    <依存関係>
        <グループID>mysql</グループID>
        <artifactId>mysql-コネクタ-java</artifactId>
    </依存関係> 
 
    <依存関係>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-boot-starter</artifactId>
        <バージョン>3.3.1</バージョン>
    </依存関係> 
 
    <依存関係>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-extension</artifactId>
        <バージョン>3.3.1</バージョン>
    </依存関係> 
 
    <依存関係>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>テスト</scope>
    </依存関係>
 </依存関係>

構成項目は次のとおりです。

サーバ：
  ポート: 9091 
 
 
春：
  データソース:
    ドライバークラス名: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/test?useUnicode= true &characterEncoding=UTF-8&useSSL= false &serverTimezone=Asia/Shanghai
    ユーザー名: root
パスワード: a123 
 
ログ記録:
レベル：
    ルート:情報

データベースフィールドをマップするために使用されるエンティティクラスは次のとおりです。

パッケージ com.yang.lock1.entity; 
 
 com.baomidou.mybatisplus.annotation.IdType をインポートします。
 com.baomidou.mybatisplus.annotation.TableField をインポートします。
 com.baomidou.mybatisplus.annotation.TableId をインポートします。
 com.baomidou.mybatisplus.annotation.TableName をインポートします。
インポート lombok.AllArgsConstructor;
 lombok.Data をインポートします。
 lombok.NoArgsConstructor をインポートします。 
 
 java.util.Dateをインポートします。 
 
 /**
 * @著者 qcy
 *@作成2020/08/25 15:03:47
 */
 @データ
@NoArgsコンストラクタ
@テーブル名(値 = "t_lock" )
パブリッククラスLock { 
 
    /**
     * 自己増加シーケンス番号
     */
    @TableId(値 = "id" 、タイプ = IdType.AUTO)
    プライベート整数ID; 
 
    /**
     * ロック名
     */
    プライベート文字列名; 
 
    /**
     * 生存時間（ミリ秒）
     */
    プライベートint生存時間; 
 
    /**
     * ロックが作成された時刻
     */
    プライベート日付createTime; 
 
    /**
     * スレッド名
     */
    プライベート文字列スレッド名;
 }

道層:

パッケージ com.yang.lock1.dao; 
 
 com.baomidou.mybatisplus.core.mapper.BaseMapper をインポートします。
 com.yang.lock1.entity.Lock をインポートします。
 org.apache.ibatis.annotations.Mapper をインポートします。 
 
 /**
 * @著者 qcy
 *@作成2020/08/25 15:06:24
 */
 @マッパー
パブリックインターフェースLockDaoはBaseMapper<Lock>を拡張します。
 }

サービスインターフェース層:

パッケージ com.yang.lock1.service; 
 
 com.baomidou.mybatisplus.extension.service.IService をインポートします。
 com.yang.lock1.entity.Lock をインポートします。 
 
 /**
 * @著者 qcy
 *@作成2020/08/25 15:07:44
 */
パブリックインターフェースLockServiceはIService<Lock>を拡張します。 
 
    /**
     * 分散ロックのブロック取得
     *
     * @param nameロック名
     * @param survivalTime 生存時間
     */
    void lock(String name , int survivalTime); 
 
    /**
     * ロックを解除する
     *
     * @param nameロック名
     */
パブリックvoid unLock(文字列名);
 }

サービス実装層:

パッケージ com.yang.lock1.service.impl; 
 
 com.baomidou.mybatisplus.extension.service.impl.ServiceImpl をインポートします。
 com.yang.lock1.dao.LockDao をインポートします。
 com.yang.lock1.entity.Lock をインポートします。
 com.yang.lock1.service.LockService をインポートします。
 lombok.extern.slf4j.Slf4j をインポートします。
 org.springframework.dao.DuplicateKeyException をインポートします。
 org.springframework.stereotype.Service をインポートします。 
 
 java.util.Dateをインポートします。 
 
 /**
 * @著者 qcy
 *@作成2020/08/25 15:08:25
 */
翻訳者
@サービス
パブリッククラス LockServiceImpl は ServiceImpl<LockDao, Lock> を拡張し、LockService を実装します { 
 
    @オーバーライド
パブリックvoid lock(String name , int survivalTime) {
        文字列 threadName = "system1-" + Thread.currentThread().getName();
        （真）の間{
            ロック lock = this.lambdaQuery().eq(Lock::getName, name ).one();
            ロック == nullの場合
                //説明 ロックなし
                ロックlk = 新しいロック();
                lk.setName(名前);
                lk.setSurvivalTime(生存時間);
                lk.setThreadName(スレッド名);
                試す {
                    保存(lk);
                    log.info(threadName + "ロックが正常に取得されました" );
戻る;
                } キャッチ (DuplicateKeyException e) {
                    //再試行を続ける
                    log.info(threadName + "ロックの取得に失敗しました" );
続く;
                }
            } 
 
            //この時点でロックが存在するので、ロックが期限切れかどうかを判断します
現在の日付= 新しい日付();
日付expireDate = new Date (lock.getCreateTime().getTime() + lock.getSurvivalTime());
            有効期限が現在より前の場合
                //ロックの有効期限が切れました
                ブール値の結果 = removeById(lock.getId());
                if (結果) {
                    log.info(threadName + "期限切れのロックを削除しました" );
                } 
 
                //ロックを取得しようとする
                ロックlk = 新しいロック();
                lk.setName(名前);
                lk.setSurvivalTime(生存時間);
                lk.setThreadName(スレッド名);
                試す {
                    保存(lk);
                    log.info(threadName + "ロックが正常に取得されました" );
戻る;
                } キャッチ (DuplicateKeyException e) {
                    log.info(threadName + "ロックの取得に失敗しました" );
                }
            }
        } 
 
    } 
 
    @オーバーライド
パブリックvoid unLock(文字列名) {
        //ロックを解除するときは、作成したロックのみを解除できることに注意する必要があります
        文字列 threadName = "system1-" + Thread.currentThread().getName();
        ロック lock = lambdaQuery().eq(Lock::getName, name ).eq(Lock::getThreadName, threadName).one();
        if (ロック != null ) {
            ブール値 b = removeById(lock.getId());
            (b) の場合 {
                log.info(threadName + "ロックを解除しました" );
            }それ以外{
                log.info(threadName + "ロックを解除する準備はできましたが、ロックの有効期限が切れており、他のクライアントによって強制的に解除されました" );
            }
        }それ以外{
            log.info(threadName + "ロックを解除する準備はできましたが、ロックの有効期限が切れており、他のクライアントによって強制的に解除されました" );
        }
    } 
 
 }

テストクラスは次のとおりです。

パッケージ com.yang.lock1; 
 
 com.yang.lock1.service.LockService をインポートします。
 lombok.extern.slf4j.Slf4j をインポートします。
 org.junit.Test をインポートします。
 org.junit.runner.RunWith をインポートします。
 org.springframework.boot.test.context.SpringBootTest をインポートします。
 org.springframework.test.context.junit4.SpringRunner をインポートします。 
 
 javax.annotation.Resource をインポートします。 
 
 /**
 * @著者 qcy
 *@作成2020/08/25 15:10:54
 */
翻訳者
SpringRunner クラスで実行します。
 @SpringBootテスト
パブリッククラスLock1ApplicationTest { 
 
    @リソース
    ロックサービスロックサービス; 
 
    @テスト
パブリックボイドテストロック（）{
        log.info( "system1はロックを取得する準備ができています" );
        lockService.lock( "キー" 、 6 * 1000);
        試す {
            //シミュレーション業務時間
            スレッドをスリープ状態にします(4 * 1000);
        } キャッチ (例外 e) {
            e.printStackTrace();
        ついに
            lockService.unLock( "キー" );
        }
    } 
 
 }

コードをコピーし、system1 を system2 に変更します。次に、両方のシステムを起動します。

system1 の出力は次のとおりです。

system2 の出力は次のとおりです。

23.037 秒に、system1 はロックの取得を試み、23.650 秒に成功し、分散ロックを保持します。 26 秒目に、system2 はロックを取得しようとしますが、ブロックされます。 27.701秒に、system1はロックを解除しました。 System2 は 27.749 秒でロックを取得し、31 秒でロックを解除しました。

ここで、system1 のサービス期間を 10 秒に変更し、system2 が system1 のタイムアウトロックを解除するシナリオをシミュレートできます。

まずシステム1を起動し、次にシステム2を起動します

このとき、system1 の出力は次のようになります。

system2 の出力は次のとおりです。

14 秒で、システム 1 がロックを取得し、その後、ビジネスは予想よりも時間がかかり、10 秒間実行する必要がありました。この間に、system1 によって作成されたロックの有効期間が終了しました。このとき、system2 は 19 秒で期限切れのロックを削除し、その後ロックを取得しました。 24 秒後、システム 1 は振り返って、ロックが解除されたことに気づきました。最後に、system2 は正常にロックを解除しました。

データベースに基づく分散ロックの実装、および悲観的ロックと楽観的ロックの方法について、別の記事を書く予定です。

<<: ビッグデータとクラウドコンピューティングの深い統合はどのような側面に反映されていますか?

>>: Kubernetes ベースのハイブリッドクラウドの長所と短所