APIにおけるレートリミティングとスロットリングは、リソースの過剰使用を防ぐための重要な設計要素です。この記事では、Pythonでこれらの制御手法をどのように実装するか、具体的なコード例とその解説、そして応用例について詳しく説明します。
目次
はじめに：レートリミティングとスロットリングとは
レートリミティングとスロットリングはAPIの利用制限に使われるテクニックです。レートリミティングは特定の時間内にAPIへのリクエスト数を制限するものであり、スロットリングはリクエストの処理速度そのものを制限するものです。
なぜ必要なのか
これらは、サービスが過負荷になることを防ぐだけでなく、悪意のある利用からも守る役割を果たします。例えば、DDoS攻撃やリソースの枯渇を防ぐためにも重要です。
Pythonでの基本的な実装方法
レートリミティングの基本的なコード
from time import sleep, time
def rate_limited(max_per_second):
min_interval = 1.0 / float(max_per_second)
def decorate(func):
last_called = [0.0]
def rate_limited_function(*args, **kwargs):
elapsed = time() - last_called[0]
if elapsed < min_interval:
sleep(min_interval - elapsed)
last_called[0] = time()
return func(*args, **kwargs)
return rate_limited_function
return decorate
@rate_limited(2) # 1秒あたり2回まで
def print_num(num):
print(num)
# テスト
for i in range(10):
print_num(i)
このコードでは、デコレータ`@rate_limited`を使って関数の呼び出し頻度を制限しています。引数には、1秒あたりの最大呼び出し回数を指定します。
スロットリングの基本的なコード
import time
def throttle(func):
min_interval = 2.0
last_called = [0.0]
def wrapper(*args, **kwargs):
elapsed = time.time() - last_called[0]
to_wait = min_interval - elapsed
if to_wait > 0:
time.sleep(to_wait)
ret = func(*args, **kwargs)
last_called[0] = time.time()
return ret
return wrapper
@throttle
def print_hello(message):
print(message)
# テスト
print_hello("こんにちは、世界")
print_hello("再度、こんにちは")
この例では、関数の実行に時間間隔を設定する`@throttle`デコレータを使用しています。ここでは2秒以上空けるようにしています。
応用例
複数ユーザーに対するレートリミティング
from collections import defaultdict
user_last_called = defaultdict(int)
def rate_limit_user(user_id, max_per_second=1):
min_interval = 1.0 / float(max_per_second)
elapsed = time() - user_last_called[user_id]
if elapsed < min_interval:
sleep(min_interval - elapsed)
user_last_called[user_id] = time()
# 使用例
for i in range(10):
rate_limit_user("user1")
print(f"User1: {i}")
この応用例では、特定のユーザーに対してレートリミティングを適用しています。
動的スロットリング
dynamic_min_interval = 1.0
def dynamic_throttle(func):
def wrapper(*args, **kwargs):
global dynamic_min_interval
elapsed = time.time() - last_called[0]
to_wait = dynamic_min_interval - elapsed
if to_wait > 0:
time.sleep(to_wait)
ret = func(*args, **kwargs)
last_called[0] = time.time()
return ret
return wrapper
@dynamic_throttle
def print_dynamic(message):
print(message)
# 使用例
dynamic_min_interval = 3.0
print_dynamic("動的にスロットリング")
この応用例では、動的にスロットリングの間隔を変更できるようにしています。
まとめ
PythonでのAPIのレートリミティングとスロットリングには多くの設計パターンがありますが、基本的な実装方法から応用例までを解説しました。これらのテクニックを活用して、APIをより堅牢で信頼性の高いものにしましょう。
コメント