เมื่อคุณสร้างแอปพลิเคชันที่พึ่งพา API จากภายนอก ไม่ว่าจะเป็น AI API หรือบริการคลาวด์ต่างๆ สิ่งที่หลีกเลี่ยงไม่ได้คือ ข้อผิดพลาดจากการเรียกใช้เกินขีดจำกัด บทความนี้จะพาคุณเข้าใจกลยุทธ์ Rate Limiting ทั้ง 4 แบบ พร้อมโค้ด Python ที่นำไปใช้งานได้จริง
สถานการณ์ข้อผิดพลาดจริง
ผมเคยพัฒนาระบบ Auto-Reply สำหรับ Line Official Account โดยใช้ AI API จากผู้ให้บริการหลายราย ช่วงแรกของการ launch ระบบทำงานได้ดี แต่หลังจากผู้ใช้งานเพิ่มขึ้นเรื่อยๆ ปัญหาเริ่มตามมา:
# ข้อผิดพลาดที่พบบ่อยเมื่อไม่มี Rate Limiting
429 Too Many Requests - Rate limit exceeded
ConnectionError: timeout after 30 seconds
HTTP 503 Service Unavailable
RateLimitError: You exceeded your current quota
หลังจากวิเคราะห์ปัญหา ผมพบว่าเราเรียก API มากเกินไปในช่วงเวลาสั้นๆ ทำให้โดน block จากผู้ให้บริการ การแก้ปัญหาคือต้อง implement Rate Limiting อย่างถูกต้อง
Rate Limiting คืออะไร และทำไมต้องสนใจ
Rate Limiting คือกลไกควบคุมจำนวน request ที่ระบบยอมรับได้ในช่วงเวลาที่กำหนด มีประโยชน์หลายอย่าง:
- ป้องกันการโจมตีแบบ DoS - จำกัด request จากผู้ไม่หวังดี
- รักษาเสถียรภาพระบบ - ไม่ให้ server ล่มจาก load สูงเกิน
- ประหยัดค่าใช้จ่าย - ควบคุมการใช้งาน API ตาม plan ที่ซื้อไว้
- 公平分配ทรัพยากร - ให้ทุกผู้ใช้เข้าถึงบริการได้อย่างเท่าเทียม
4 กลยุทธ์ Rate Limiting ที่ต้องรู้จัก
1. Fixed Window (Sliding Counter)
วิธีที่ง่ายที่สุด คือนับ request ในช่วงเวลาคงที่ เช่น 100 request ต่อ 1 นาที เมื่อครบแล้วต้องรอนาทีถัดไปจึงเริ่มนับใหม่
"""
Fixed Window Rate Limiter - กลยุทธ์ที่ง่ายที่สุด
หลักการ: นับ request ในช่วงเวลาที่กำหนด (window) แล้วรีเซ็ตเมื่อหมดเวลา
"""
import time
from datetime import datetime, timedelta
from collections import defaultdict
class FixedWindowRateLimiter:
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = defaultdict(list) # {client_id: [timestamps]}
def _clean_old_requests(self, client_id: str, current_time: float):
"""ลบ request ที่เก่ากว่า window"""
cutoff = current_time - self.window_seconds
self.requests[client_id] = [
ts for ts in self.requests[client_id]
if ts > cutoff
]
def is_allowed(self, client_id: str) -> tuple[bool, dict]:
"""
ตรวจสอบว่า request นี้ได้รับอนุญาตหรือไม่
Returns: (is_allowed, info_dict)
"""
current_time = time.time()
self._clean_old_requests(client_id, current_time)
current_count = len(self.requests[client_id])
remaining = self.max_requests - current_count
if current_count < self.max_requests:
self.requests[client_id].append(current_time)
return True, {
"allowed": True,
"remaining": remaining - 1,
"reset_at": datetime.fromtimestamp(
current_time + self.window_seconds
).isoformat()
}
else:
oldest = min(self.requests[client_id])
retry_after = int(oldest + self.window_seconds - current_time)
return False, {
"allowed": False,
"remaining": 0,
"retry_after": retry_after,
"reset_at": datetime.fromtimestamp(
oldest + self.window_seconds
).isoformat()
}
ตัวอย่างการใช้งาน
if __name__ == "__main__":
# อนุญาต 5 request ต่อ 10 วินาที
limiter = FixedWindowRateLimiter(max_requests=5, window_seconds=10)
for i in range(8):
client_id = "user_001"
allowed, info = limiter.is_allowed(client_id)
print(f"Request {i+1}: {'✅' if allowed else '❌'} - {info}")
time.sleep(1) # รอ 1 วินาทีระหว่าง request
ข้อดี: ง่ายต่อการ implement และเข้าใจ
ข้อเสีย: มี "traffic spike" ได้ที่ขอบเขต window เช่น request 100 ตัว ณ นาทีที่ 59 และ 100 ตัวอีก ณ นาทีที่ 1 รวม 200 request ใน 2 นาที
2. Sliding Window Log
ปรับปรุงจาก Fixed Window โดยใช้ timestamp ของแต่ละ request แทนการนับแบบคงที่ ทำให้การกระจายตัวของ request นุ่มนวลกว่า
"""
Sliding Window Log Rate Limiter - แม่นยำกว่า Fixed Window
หลักการ: เก็บ log ของทุก request แล้วนับเฉพาะ request ที่อยู่ในช่วงเวลาปัจจุบัน
"""
import time
from datetime import datetime
from collections import deque
from threading import Lock
class SlidingWindowLogLimiter:
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.window_start = time.time()
self.request_count = 0
self.lock = Lock()
def is_allowed(self, client_id: str) -> tuple[bool, dict]:
"""
ตรวจสอบ request ด้วย sliding window algorithm
"""
current_time = time.time()
with self.lock:
# ตรวจสอบว่า window เก่าหรือยัง
elapsed = current_time - self.window_start
if elapsed >= self.window_seconds:
# รีเซ็ต window ใหม่
self.window_start = current_time
self.request_count = 0
# ตรวจสอบจำนวน request
if self.request_count < self.max_requests:
self.request_count += 1
remaining = self.max_requests - self.request_count
reset_at = self.window_start + self.window_seconds
return True, {
"allowed": True,
"remaining": remaining,
"reset_in": int(self.window_seconds - elapsed),
"reset_at": datetime.fromtimestamp(reset_at).isoformat()
}
else:
return False, {
"allowed": False,
"remaining": 0,
"retry_after": int(self.window_seconds - elapsed),
"reset_at": datetime.fromtimestamp(
self.window_start + self.window_seconds
).isoformat()
}
ทดสอบ Sliding Window vs Fixed Window
if __name__ == "__main__":
print("=" * 50)
print("เปรียบเทียบ Sliding Window vs Fixed Window")
print("=" * 50)
# Fixed Window - อนุญาต 5 request ต่อ 10 วินาที
fixed_limiter = FixedWindowRateLimiter(max_requests=5, window_seconds=10)
print("\n[Fixed Window] 8 requests ใน 8 วินาที:")
for i in range(8):
allowed, info = fixed_limiter.is_allowed("user_test")
status = "✅" if allowed else "❌"
print(f" Request {i+1}: {status} | คงเหลือ: {info['remaining']}")
time.sleep(1)
print("\n✅ สังเกตว่า request 6-8 ถูกปฏิเสธเพราะนับรวมกันได้ 5 ใน window เดียว")
3. Token Bucket
เป็นกลยุทธ์ที่นิยมใช้มากที่สุด คล้ายกับถังที่มีโทเค็นจำนวนหนึ่ง ทุกครั้งที่มี request โทเค็นจะถูกหักออก และโทเค็นจะค่อยๆ เติมกลับมาตามเวลาที่กำหนด
"""
Token Bucket Rate Limiter - กลยุทธ์ที่นิยมที่สุด
หลักการ: มี "ถัง" เก็บโทเค็น ทุก request ใช้ 1 โทเค็น โทเค็นเติมกลับมาตาม refill_rate
"""
import time
import threading
from datetime import datetime
from typing import Optional
class TokenBucketLimiter:
"""
Token Bucket Algorithm Implementation
Attributes:
capacity: จำนวนโทเค็นสูงสุดในถัง
refill_rate: จำนวนโทเค็นที่เติมต่อวินาที
tokens: โทเค็นปัจจุบันในถัง
last_refill: timestamp ของการเติมโทเค็นล่าสุด
"""
def __init__(
self,
capacity: int = 100,
refill_rate: float = 10.0 # tokens per second
):
self.capacity = capacity
self.refill_rate = refill_rate
self.tokens = float(capacity) # เริ่มต้นด้วยถังเต็ม
self.last_refill = time.time()
self.lock = threading.Lock()
def _refill(self):
"""เติมโทเค็นตามเวลาที่ผ่านไป"""
now = time.time()
elapsed = now - self.last_refill
# คำนวณโทเค็นที่ต้องเติม
tokens_to_add = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + tokens_to_add)
self.last_refill = now
def consume(self, tokens: int = 1) -> tuple[bool, dict]:
"""
พยายามใช้โทเค็น
Returns: (success, info)
"""
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True, {
"allowed": True,
"tokens_remaining": self.tokens,
"refill_rate": self.refill_rate,
"capacity": self.capacity,
"timestamp": datetime.now().isoformat()
}
else:
# คำนวณเวลาที่ต้องรอ
tokens_needed = tokens - self.tokens
wait_time = tokens_needed / self.refill_rate
return False, {
"allowed": False,
"tokens_remaining": self.tokens,
"wait_seconds": round(wait_time, 2),
"refill_rate": self.refill_rate,
"timestamp": datetime.now().isoformat()
}
def get_status(self) -> dict:
"""ดูสถานะปัจจุบันของ bucket"""
with self.lock:
self._refill()
return {
"tokens": round(self.tokens, 2),
"capacity": self.capacity,
"refill_rate": self.refill_rate,
"utilization": round(
(self.capacity - self.tokens) / self.capacity * 100, 1
)
}
ตัวอย่างการใช้งาน
if __name__ == "__main__":
print("=" * 60)
print("Token Bucket Rate Limiter Demo")
print("=" * 60)
# สร้าง limiter: ถังขนาด 5 โทเค็น, เติม 2 โทเค็น/วินาที
bucket = TokenBucketLimiter(capacity=5, refill_rate=2.0)
print("\n⏱️ ทดสอบ: ส่ง request 8 ครั้งใน 4 วินาที")
print("-" * 60)
for i in range(8):
success, info = bucket.consume()
status = "✅" if success else "❌"
tokens = info.get("tokens_remaining", 0)
wait = info.get("wait_seconds", "-")
print(f" Request {i+1}: {status} | โทเค็นเหลือ: {tokens:.1f}", end="")
if not success:
print(f" | รอ: {wait}s", end="")
print()
time.sleep(0.5) # ส่ง request ทุก 0.5 วินาที
print("\n📊 สถานะปัจจุบัน:")
print(f" {bucket.get_status()}")
print("\n💡 สังเกต: Request 6-8 ถูกปฏิเสธ เพราะโทเค็นหมด")
print(" แต่โทเค็นจะค่อยๆ เติมกลับมา ทำให้ request ต่อไปผ่านได้")
4. Leaky Bucket
ตรงข้ามกับ Token Bucket - request ที่เข้ามาจะถูก "รีด" (leak) ออกด้วยอัตราคงที่ เหมาะกับการควบคุม output rate ไม่ให้เกินขีดจำกัด
"""
Leaky Bucket Rate Limiter - ควบคุม output rate
หลักการ: request เข้ามาได้เรื่อยๆ แต่ออกไปด้วยอัตราคงที่
ถ้า bucket เต็ม request ใหม่จะถูกปฏิเสธ
"""
import time
import threading
from collections import deque
from datetime import datetime
class LeakyBucketLimiter:
"""
Leaky Bucket Algorithm
เหมาะกับ: ระบบที่ต้องการส่ง request ไปยัง API ด้วยอัตราคงที่
เช่น: batch processing, report generation
"""
def __init__(
self,
capacity: int = 100,
leak_rate: float = 10.0 # requests per second
):
self.capacity = capacity
self.leak_rate = leak_rate
self.bucket = deque()
self.last_leak_time = time.time()
self.lock = threading.Lock()
self.total_processed = 0
self.total_rejected = 0
def _leak(self):
"""รีด request ออกตามอัตราที่กำหนด"""
now = time.time()
elapsed = now - self.last_leak_time
# คำนวณจำนวน request ที่รีดได้
leak_count = int(elapsed * self.leak_rate)
if leak_count > 0 and len(self.bucket) > 0:
for _ in range(min(leak_count, len(self.bucket))):
self.bucket.popleft()
self.last_leak_time = now
def add_request(self, request_id: Optional[str] = None) -> tuple[bool, dict]:
"""
เพิ่ม request เข้า bucket
"""
request_id = request_id or f"req_{time.time()}"
with self.lock:
self._leak()
if len(self.bucket) < self.capacity:
self.bucket.append(request_id)
return True, {
"allowed": True,
"queue_size": len(self.bucket),
"capacity": self.capacity,
"leak_rate": self.leak_rate,
"request_id": request_id
}
else:
self.total_rejected += 1
return False, {
"allowed": False,
"queue_size": len(self.bucket),
"capacity": self.capacity,
"reason": "Bucket is full",
"total_rejected": self.total_rejected
}
def process_next(self) -> tuple[bool, str]:
"""ดึง request ออกมาประมวลผล (leak)"""
with self.lock:
self._leak()
if len(self.bucket) > 0:
self.total_processed += 1
request_id = self.bucket.popleft()
return True, request_id
return False, None
ตัวอย่างการใช้งาน
if __name__ == "__main__":
print("=" * 60)
print("Leaky Bucket Rate Limiter Demo")
print("=" * 60)
# ถังขนาด 10 request, leak 5 request/วินาที
leaky = LeakyBucketLimiter(capacity=10, leak_rate=5.0)
print("\n📥 เพิ่ม 20 requests ใน 2 วินาที:")
print("-" * 60)
accepted = 0
rejected = 0
for i in range(20):
success, info = leaky.add_request(f"task_{i+1}")
if success:
accepted += 1
print(f" ✅ เพิ่ม {info['request_id']} | คิว: {info['queue_size']}")
else:
rejected += 1
print(f" ❌ ถูกปฏิเสธ | {info['reason']}")
time.sleep(0.1) # เพิ่ม request ทุก 0.1 วินาที
print(f"\n📊 สรุป: รับ {accepted} requests, ปฏิเสธ {rejected} requests")
print(f" อัตรา leak: {leaky.leak_rate} requests/วินาที")
print(f" รอประมวลผล: {len(leaky.bucket)} requests")
เปรียบเทียบ 4 กลยุทธ์
| กลยุทธ์ | หลักการ | ข้อดี | ข้อเสีย | เหมาะกับ |
|---|---|---|---|---|
| Fixed Window | นับ request ในช่วงเวลาคงที่ | ง่าย, ใช้ memory น้อย | Traffic spike ที่ขอบเขต window | ระบบง่ายๆ, prototype |
| Sliding Window | นับ request แบบ continuous | แม่นยำกว่า, ไม่มี spike | ใช้ memory มากขึ้น | API ที่ต้องการความ精确 |
| Token Bucket | ใช้โทเค็น, เติมกลับตามเวลา | รองรับ burst, efficiency สูง | ซับซ้อนขึ้นเล็กน้อย | AI API, เว็บไซต์, mobile apps |
| Leaky Bucket | รีด request ออกด้วยอัตราคงที่ | output rate คงที่, smooth | request อาจต้องรอนาน | Batch processing, IoT |
Implementation กับ HolySheep AI API
สำหรับการใช้งาน AI API จริง ผมแนะนำให้ใช้ Token Bucket เพราะรองรับ burst ของ request ได้ดี แต่ยังควบคุมอัตราเฉลี่ยได้ นี่คือตัวอย่างการนำไปใช้กับ HolySheep AI:
"""
ตัวอย่างการใช้ Rate Limiter กับ HolySheep AI API
Base URL: https://api.holysheep.ai/v1
"""
import os
import time
import httpx
from token_bucket import TokenBucketLimiter
สร้าง client พร้อม rate limiting
class HolySheepAIClient:
"""
HolySheep AI API Client พร้อม built-in Rate Limiting
Rate Limits ของ HolySheep AI:
- Free tier: 60 requests/minute
- Pro tier: 600 requests/minute
- Enterprise: custom limits
"""
def __init__(
self,
api_key: str,
rate_limit: int = 60, # requests per minute
tier: str = "free"
):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.tier = tier
# ตั้งค่า rate limiter ตาม tier
if tier == "free":
max_tokens = 60
refill_rate = 1.0 # 1 request/second
elif tier == "pro":
max_tokens = 600
refill_rate = 10.0 # 10 requests/second
else: # enterprise
max_tokens = 6000
refill_rate = 100.0
self.rate_limiter = TokenBucketLimiter(
capacity=max_tokens,
refill_rate=refill_rate
)
def chat_completions(
self,
model: str = "gpt-4.1",
messages: list[dict],
max_retries: int = 3
) -> dict:
"""
ส่ง request ไปยัง HolySheep AI Chat Completions API
"""
url = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages
}
for attempt in range(max_retries):
# ตรวจสอบ rate limit ก่อนส่ง request
allowed, info = self.rate_limiter.consume()
if not allowed:
wait_time = info["wait_seconds"]
print(f"⏳ Rate limited. รอ {wait_time} วินาที...")
time.sleep(wait_time + 0.1)
continue
try:
with httpx.Client(timeout=30.0) as client:
response = client.post(url, json=payload, headers=headers)
if response.status_code == 429:
# Rate limit hit - รอแล้วลองใหม่
retry_after = int(
response.headers.get("Retry-After", 1)
)
print(f"⚠️ API rate limit. รอ {retry_after}s...")
time.sleep(retry_after)
continue
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
if attempt == max_retries - 1:
raise
print(f"❌ HTTP Error: {e}. ลองใหม่...")
time.sleep(2 ** attempt)
raise Exception("Max retries exceeded")
ตัวอย่างการใช้งาน
if __name__ == "__main__":
# ดึง API key จาก environment variable
api_key = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
# สร้าง client (Free tier)
client = HolySheepAIClient(
api_key=api_key