サーキットブレーカーと同時処理数の制限
失敗する呼出しを止める条件と、受け付けた処理をどれだけ保持するかを分けて設計します。
説明
下流が失敗しているときに同じ要求を送り続けると、待機や再試行で上流の資源まで埋まる場合があります。サーキットブレーカーは失敗が続く呼出しを一時的に止め、回復を少数の試行で確認します。同時実行数や待機数の上限は、別の仕組みとして必要です。
3回の連続失敗で5秒間遮断する固定ブレーカーを使い、CLOSED・OPEN・HALF_OPENを進めます。その後、実asyncioの8要求で、無制限に近い実行、同時数2だけの制限、待機数3も加えた制限を比較します。
失敗する呼出しを止める条件と、受け付けた処理をどれだけ保持するかを分けて設計します。
1通常の呼出しと失敗回数
CLOSEDは呼出しを通す状態です。最初の失敗を発生させます。healthy=Falseは、この固定の下流呼出しが失敗する条件です。
from protection import Breaker, burst
import asyncio
breaker = Breaker()3回失敗・5秒遮断の状態を用意する
breaker.call(False)最初の失敗を記録するfailures=1、state=CLOSEDです。1回の失敗だけでは遮断せず、この例では連続3回を閾値にしています。startedは実際に下流側の処理へ進んだ回数です。
2三回の失敗で遮断へ移る
同じ失敗をあと2回起こします。
breaker.call(False)連続失敗数を閾値まで進める
breaker.call(False)3回目でOPEN、retry_at=5になります。以後は待機期限まで下流呼出しを通しません。名前のOPENは回路を開いて通さない状態を表します。
3遮断中は下流へ送らない
時刻0のまま追加の要求を2回与えます。
breaker.call(False)OPEN中の追加要求を評価する
breaker.call(False)どちらもrejectedで、startedは3のままです。失敗しそうな下流へ新しい処理を送り続けず、呼出し側へ早く失敗を返す状態です。
4待機後の試行が失敗する場合
時刻6へ進め、一回の試行を行います。
breaker.clock = 6最初の遮断期限を過ぎる時刻へ進める
breaker.call(False)回復確認の試行を失敗させるHALF_OPENで試し、失敗したので再びOPEN、retry_at=11です。時間が経過したことだけで正常と判断せず、実際の試行結果を使います。
5回復確認の成功で通常へ戻る
時刻12で成功する試行を行い、その後もう一度成功させます。
breaker.clock = 12再遮断の期限を過ぎる時刻へ進める
breaker.call(True)成功する試行と通常呼出しを行う
breaker.call(True)試行の成功でCLOSEDへ戻り、失敗数をリセットします。次の呼出しも通常どおり通ります。実際には、試行の同時数や必要な連続成功数も方針として決めます。
6八件を同時に動かす場合
次は別の問題として、短い非同期処理を8件まとめて実行します。待機は実asyncio.sleepで行い、同時に動いた最大数を数えます。
unbounded = asyncio.run(burst(8))最大8件を並行して実行する8件すべて完了し、peak_active=8です。下流が正常でも、同時に実行する件数が増えると接続やメモリなどを占有します。
7同時数だけを二件へ制限
Semaphoreで同時実行を2にします。受付件数には上限を加えていません。
limited = asyncio.run(burst(2))同時実行2件、受付8件で動かすpeak_active=2ですが、8件すべてを受け付け、6件が待機する構成です。同時実行だけを制限しても、待機する要求が際限なく増える問題は別に残ります。
8待機できる件数にも上限を置く
実行2件と待機3件、合計5件まで受け付ける条件にします。8件は同時に到着した条件です。
bounded = asyncio.run(burst(2, queue_limit=3))同時実行と待機の両方に上限を置く5件が完了し3件を拒否、peak_active=2、admitted_waiters=3です。拒否した要求をどう再試行・通知するかも含めて、処理能力に合う受付方針を決めます。
まとめ
ブレーカーは失敗状態を見て呼出しを遮断し、限られた試行で回復を確認します。
同時実行数の制限だけでは待機数を制限したことになりません。
処理能力を超えた要求の受付・拒否・再試行を合わせて設計します。
操作履歴と解説を振り返れます。
Python 3
from protection import Breaker, burst
import asyncio
breaker = Breaker()
breaker.call(False)result=failure state=CLOSED failures=1 started=1 retry_at=0
breaker.call(False)
breaker.call(False)result=failure state=CLOSED failures=2 started=2 retry_at=0 result=failure state=OPEN failures=3 started=3 retry_at=5
breaker.call(False)
breaker.call(False)state=OPEN rejected without downstream call started=3 state=OPEN rejected without downstream call started=3
breaker.clock = 6
breaker.call(False)probe state=HALF_OPEN result=failure state=OPEN failures=4 started=4 retry_at=11
breaker.clock = 12
breaker.call(True)
breaker.call(True)probe state=HALF_OPEN result=success state=CLOSED started=5 result=success state=CLOSED started=6
unbounded = asyncio.run(burst(8))requests=8 completed=8 rejected=0 peak_active=8 admitted_waiters=0
limited = asyncio.run(burst(2))requests=8 completed=8 rejected=0 peak_active=2 admitted_waiters=6
bounded = asyncio.run(burst(2, queue_limit=3))requests=8 completed=5 rejected=3 peak_active=2 admitted_waiters=3
演習完了