Go 分散トークンバケットの電流制限 + 最終利益の保証

この記事は、Ouyang An が執筆した WeChat パブリックアカウント「Microservice Practice」から転載したものです。この記事を転載する場合は、Microservice Practice のパブリックアカウントにお問い合わせください。

前の記事では、固定時間ウィンドウの電流制限では突然の要求のピークに対応できないことを説明しました。この記事で説明するトークンバケット回路アルゴリズムは、このシナリオをより適切に処理できます。

仕組み

単位時間あたりに生成されるトークンは、バケット容量の上限に達するまで一定の割合でバケットに入れられます。

リクエストを処理します。そのたびに 1 つ以上のトークンを取得しようとします。取得できた場合はリクエストを処理し、取得できなかった場合はリクエストを拒否します。

長所と短所

アドバンテージ

瞬間的なバーストトラフィックを効果的に処理でき、バケット内のトークンをトラフィックバッファとして使用してバーストトラフィックをスムーズに処理できます。

欠点

実装はより複雑です。

コードの実装

コア/制限/トークン制限.go

分散環境でバケットとトークンのストレージコンテナーとして Redis を使用することを検討し、Lua スクリプトを使用してアルゴリズムプロセス全体を実装します。

Redis Lua スクリプト

-- 1秒あたりに生成されるトークンの数、つまりトークン生成速度 
ローカルレート = tonumber(ARGV[1])
 -- バケット容量 
ローカル容量 = tonumber(ARGV[2])
 -- 現在のタイムスタンプ 
ローカルnow = tonumber(ARGV[3])
 -- 現在のリクエストトークン番号 
ローカル要求 = tonumber(ARGV[4])
 -- バケツを満たすのに何秒かかりますか 
ローカルfill_time = 容量/レート
-- 切り捨て、ttl は fill 時間の 2 倍になります 
ローカルttl = math.floor(fill_time*2)
 -- 現在のタイムバケット容量 
ローカルlast_tokens = tonumber(redis.call( "get" , KEYS[1]))
 -- 現在のバケット容量が0の場合、初めて入力することを意味し、デフォルトの容量はバケットの最大容量です。  
 last_tokens == nilの場合 
 last_tokens = 容量
終わり 
 -- 最終更新時刻 
ローカルlast_refreshed = tonumber(redis.call( "get" , KEYS[2]))
 -- 最初のエントリの更新時間を 0 に設定します 
 last_refreshed == nilの場合 
最終更新日 = 0
終わり 
 -- 最後のリクエストからの経過時間 
ローカルデルタ = 数学。最大(0, 現在 - 最終更新日)
 -- 最後のリクエストからの時間間隔、生成できるトークンの総数、最大容量を超えた場合、超過したトークンは破棄されます 
ローカルfilled_tokens = math.最小(容量、last_tokens+(delta*rate))
 -- このリクエストのトークンの数は十分ですか?  
ローカルで許可 = filled_tokens >= 要求
-- 残りのバケット数 
ローカルnew_tokens = filled_tokens
 -- このトークンの適用を許可し、残りの数量を計算します 
許可されれば 
 new_tokens = filled_tokens - リクエスト済み
終わり 
 -- 残りのトークンの数を設定する 
 redis.call( "setex" , KEYS[1], ttl, new_tokens)
 -- 更新時間を設定する 
 redis.call( "setex" , KEYS[2], ttl, now) 
 
返品可能

トークンバケットリミッターの定義

TokenLimiter構造体型{
    // 1秒あたりの生産率
    レートint  
    // バケット容量
    バースト整数 
    // ストレージコンテナ
    *redis.Redis を保存します。
    // Redisキー 
    トークンキー文字列
    // バケット更新時間キー 
    タイムスタンプキー文字列
    // ロック
    rescueLock sync.Mutex
    // Redis ヘルスフラグ
    redisAlive uint32
    // Redis が失敗した場合は、インプロセス トークン バケットの現在のリミッターを使用します
    レスキューリミッター *xrate.リミッター
    // Redis 監視検出タスク識別子
    monitorStarted ブール値
} 
 
 func NewTokenLimiter(rate, burst int , store *redis.Redis, key string) *TokenLimiter {
    tokenKey := fmt.Sprintf(tokenFormat, key )
    timestampKey := fmt.Sprintf(timestampFormat, key ) 
 
 &TokenLimiter{を返す
        レート: レート、
        バースト：バースト、
        ストア: ストア、
        トークンキー: トークンキー、
        タイムスタンプキー: タイムスタンプキー、
        ライブ: 1,
        レスキューリミッター: xrate.NewLimiter(xrate.Every( time . Second / time .Duration(rate)), burst),
    }
 }

トークンを取得する

func (lim *TokenLimiter) reserveN(now time . Time , n int ) bool {
    // Redis が正常かどうかを判定する
    // Redis が失敗したときにプロセス内電流制限を使用する
    // ボトムライン保護
    atomic.LoadUint32(&lim.redisAlive) == 0 の場合 {
 lim.rescueLimiter.AllowN(now, n)を返します。
    }
    // スクリプトを実行してトークンを取得します
    応答、err := lim.store.Eval(
        スクリプト、
        []弦{
            limit.tokenKey、
            lim.timestampKey、
        },
        []弦{
            strconv.Itoa(制限率)、
            strconv.Itoa(limit.burst)、
            strconv.FormatInt(now.Unix(), 10)、
            strconv.Itoa(n)、
        })
    // redis 許可 == false  
    // Lua ブール値false -> r Nil 一括返信
    //キーが存在しないケースの特別な処理
    err == redis.Nilの場合{
戻る 間違い 
    }そうでなければ err != nil {
        logx.Errorf( "レートリミッターの使用に失敗しました: %s、レスキューにはインプロセスリミッターを使用してください" , err)
        // 実行例外、Redis ヘルス検出タスクを開始
        // フォールバックとしてインプロセス電流リミッターも使用する
        lim.startMonitor()
 lim.rescueLimiter.AllowN(now, n)を返します。
    } 
 
    コード、ok := resp.(int64)
    !okの場合{
        logx.Errorf( "redis スクリプトの評価に失敗しました: %v、レスキューにはインプロセス リミッターを使用してください" , resp)
        lim.startMonitor()
 lim.rescueLimiter.AllowN(now, n)を返します。
    } 
 
    // redis 許可 == true  
    // Lua ブール値true -> r整数応答（値1 ）
戻りコード == 1
 }

Redis 障害フォールバック戦略

バックアップ戦略の設計は非常に詳細です。 redis が利用できない場合、基本的なフロー制限が利用可能であり、サービスが過負荷にならないようにするために、ratelimit のスタンドアロンバージョンがバックアップフロー制限として起動されます。

 // Redis のヘルス検出を有効にする
func (lim *TokenLimiter) startMonitor() {
    lim.rescueLock.Lock()
    lim.rescueLock.Unlock() を延期する
    // 繰り返し開くのを防ぐ
    lim.monitorStartedの場合{
戻る 
    } 
 
    // タスクとヘルスフラグを設定する
    lim.monitorStarted = true  
    アトミック.StoreUint32(&lim.redisAlive, 0)
    // ヘルス検出
    lim.waitForRedis() を実行する
} 
 
 // Redis ヘルス検出スケジュールタスク
func (lim *TokenLimiter) waitForRedis() {
    ティッカー:=時間.NewTicker(pingInterval)
    // この関数はヘルス検出が成功したときにコールバックされます
    遅延関数() {
        ticker.Stop()
        lim.rescueLock.Lock()
        lim.monitorStarted = false  
        lim.rescueLock.Unlock()
    }() 
 
範囲ticker.C {
        // pingはRedisの組み込みヘルス検出コマンドです
        lim.store.Ping() の場合 {
            //ヘルス検出が成功したので、ヘルスフラグを設定します
            アトミック.StoreUint32(&limit.redisAlive, 1)
戻る 
        }
    }
 }