สวัสดีครับ ผมเป็นวิศวกรที่เคยเจอปัญหา API ล่มเพราะโค้ดไม่ได้จำกัดจำนวนคำขอมาหลายครั้ง วันนี้จะมาสอนคุณสร้างระบบ Rate Limiter แบบ Token Bucket ตั้งแต่เริ่มต้น อธิบายง่ายๆ ไม่ต้องมีพื้นฐานก็เข้าใจได้

ทำไมต้องมี Rate Limiter?

ลองนึกภาพร้านกาแฟที่มีพนักงาน 1 คน แต่มีลูกค้าพร้อมกัน 100 คน ร้านก็จะล่ม ระบบ API ก็เช่นเดียวกัน ถ้าไม่จำกัดจำนวนคำขอ จะเกิดปัญหา:

ระบบ Rate Limiter จะคอยจำกัดว่าผู้ใช้หรือแอปจะส่งคำขอได้กี่ครั้งต่อวินาที เหมือนประตูหมุดที่ควบคุมจำนวนคนเข้าออก

Token Bucket Algorithm ทำงานอย่างไร?

ถ้าให้ผมอธิบายแบบเข้าใจง่าย: ลองนึกถึงถังน้ำที่มีท่อน้ำเข้าและท่อน้ำออก

วิธีนี้ดีกว่าการนับคำขอตรงๆ เพราะช่วยให้:

เริ่มสร้าง Token Bucket Rate Limiter กันเลย

ในการทดลองนี้ ผมจะใช้ HolySheep AI เป็น API หลัก เพราะราคาถูกมาก (GPT-4.1 เพียง $8/ล้าน Token, DeepSeek V3.2 ราคาเบาๆ $0.42/ล้าน Token) และความเร็วต่ำกว่า 50 มิลลิวินาที ช่วยให้ทดลองได้สะดวกโดยไม่ต้องกังวลเรื่องค่าใช้จ่าย

ขั้นตอนที่ 1: สร้าง Rate Limiter แบบง่ายที่สุด

import time
from threading import Lock

class TokenBucketRateLimiter:
    """
    ระบบควบคุมจำนวนคำขอแบบ Token Bucket
    สำหรับมือใหม่ที่ไม่มีประสบการณ์ API
    """
    
    def __init__(self, capacity=10, refill_rate=5):
        """
        capacity: จำนวน Token สูงสุดที่ถังเก็บได้
        refill_rate: จำนวน Token ที่เติมต่อวินาที
        """
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = capacity  # เริ่มต้นถังเต็ม
        self.last_refill = time.time()
        self.lock = Lock()
    
    def _refill(self):
        """เติม Token ตามเวลาที่ผ่านไป"""
        now = time.time()
        elapsed = now - self.last_refill
        new_tokens = elapsed * self.refill_rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_refill = now
    
    def allow_request(self):
        """
        ตรวจสอบว่าคำขอนี้ผ่านได้หรือไม่
        คืนค่า True = ผ่าน, False = ถูกบล็อก
        """
        with self.lock:
            self._refill()
            if self.tokens >= 1:
                self.tokens -= 1
                return True
            return False
    
    def wait_and_request(self):
        """
        รอจนกว่าจะมี Token แล้วค่อยส่งคำขอ
        เหมาะสำหรับงานที่ต้องการความแน่นอน
        """
        while True:
            if self.allow_request():
                return True
            time.sleep(0.1)  # รอ 0.1 วินาทีแล้วลองใหม่


ทดสอบการทำงาน

limiter = TokenBucketRateLimiter(capacity=5, refill_rate=2)

print("ทดสอบ Rate Limiter ด้วยคำขอ 10 ครั้ง")
for i in range(10):
    result = limiter.allow_request()
    status = "ผ่าน ✅" if result else "ถูกบล็อก ⛔"
    print(f"คำขอที่ {i+1}: {status} | Token คงเหลือ: {limiter.tokens:.2f}")
    time.sleep(0.5)

ผลลัพธ์ที่ได้จะเป็นดังนี้:

ทดสอบ Rate Limiter ด้วยคำขอ 10 ครั้ง
คำขอที่ 1: ผ่าน ✅ | Token คงเหลือ: 4.00
คำขอที่ 2: ผ่าน ✅ | Token คงเหลือ: 3.00
คำขอที่ 3: ผ่าน ✅ | Token คงเหลือ: 2.00
คำขอที่ 4: ผ่าน ✅ | Token คงเหลือ: 1.00
คำขอที่ 5: ผ่าน ✅ | Token คงเหลือ: 0.00
คำขอที่ 6: ถูกบล็อก ⛔ | Token คงเหลือ: 0.00
คำขอที่ 7: ถูกบล็อก ⛔ | Token คงเหลือ: 1.00
คำขอที่ 8: ผ่าน ✅ | Token คงเหลือ: 0.00
คำขอที่ 9: ถูกบล็อก ⛔ | Token คงเหลือ: 1.00
คำขอที่ 10: ผ่าน ✅ | Token คงเหลือ: 0.00

สังเกตว่าคำขอที่ 7 เริ่มผ่านได้เพราะระบบเติม Token มาแล้ว 2 วินาที × 2 Token/วินาที = 4 Token แต่เนื่องจากถังจุดได้แค่ 5 จึงเหลือ 1 Token

ขั้นตอนที่ 2: เชื่อมต่อกับ API จริง (HolySheep AI)

ต่อไปเราจะนำ Rate Limiter ไปใช้กับ API จริง ในที่นี้ผมจะใช้ HolySheep AI เพราะมี API ที่เสถียรและราคาถูก รองรับหลายโมเดล (GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2)

import requests
import time

class APIClientWithRateLimit:
    """
    ไคลเอนต์สำหรับเรียก API แบบมี Rate Limiter ในตัว
    ออกแบบมาสำหรับ HolySheep AI โดยเฉพาะ
    """
    
    def __init__(self, api_key, requests_per_second=10):
        """
        api_key: คีย์ API จาก HolySheep AI
        requests_per_second: จำนวนคำขอต่อวินาทีที่อนุญาต
        """
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"  # URL ของ HolySheep
        self.limiter = TokenBucketRateLimiter(
            capacity=requests_per_second,
            refill_rate=requests_per_second
        )
    
    def chat_completion(self, messages, model="gpt-4.1"):
        """
        ส่งข้อความไปยัง AI แล้วรอคำตอบ
        
        messages: รายการข้อความ [{"role": "...", "content": "..."}]
        model: ชื่อโมเดล (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2)
        """
        # รอจนกว่า Rate Limiter อนุญาต
        self.limiter.wait_and_request()
        
        # ส่งคำขอไปยัง API
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            
            if response.status_code == 200:
                return response.json()
            elif response.status_code == 429:
                print("⚠️ เกินขีดจำกัด API ของเซิร์ฟเวอร์ รอสักครู่...")
                time.sleep(5)  # รอ 5 วินาทีแล้วลองใหม่
                return self.chat_completion(messages, model)
            else:
                raise Exception(f"API Error: {response.status_code} - {response.text}")
                
        except requests.exceptions.Timeout:
            raise Exception("คำขอหมดเวลา กรุณาลองใหม่")
        except requests.exceptions.ConnectionError:
            raise Exception("ไม่สามารถเชื่อมต่อเซิร์ฟเวอร์ได้")


ตัวอย่างการใช้งาน

if __name__ == "__main__":
    # ใส่ API Key ของคุณที่นี่
    API_KEY = "YOUR_HOLYSHEEP_API_KEY"
    
    client = APIClientWithRateLimit(
        api_key=API_KEY,
        requests_per_second=5  # จำกัด 5 คำขอต่อวินาที
    )
    
    messages = [
        {"role": "system", "content": "คุณเป็นผู้ช่วยที่เป็นมิตร"},
        {"role": "user", "content": "อธิบาย Token Bucket แบบเข้าใจง่าย"}
    ]
    
    print("กำลังส่งคำขอไปยัง API...")
    start_time = time.time()
    
    result = client.chat_completion(messages, model="deepseek-v3.2")
    
    elapsed = (time.time() - start_time) * 1000
    print(f"✅ ได้รับคำตอบแล้ว (ใช้เวลา {elapsed:.0f} มิลลิวินาที)")
    print(f"คำตอบ: {result['choices'][0]['message']['content']}")

ขั้นตอนที่ 3: ระบบแบบกระจาย (Distributed Rate Limiter)

ในโลกจริง เรามีหลายเซิร์ฟเวอร์ทำงานพร้อมกัน ถ้าใช้ Rate Limiter แบบเดิมแต่ละเซิร์ฟเวอร์จะนับ Token แยกกัน ทำให้ผู้ใช้ส่งคำขอได้มากกว่าที่กำหนด

วิธีแก้คือใช้ Redis เป็นที่เก็บ Token กลาง ให้ทุกเซิร์ฟเวอร์มาเช็คที่เดียวกัน

import redis
import time
import json

class DistributedTokenBucket:
    """
    Rate Limiter แบบกระจายใช้ Redis เป็นตัวเก็บ Token กล

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง