ในฐานะวิศวกรที่ดูแลระบบที่เชื่อมต่อกับ Exchange API หลายตัว ผมเคยเจอปัญหา 429 Too Many Requests ที่ทำให้ระบบหยุดชะงักอยู่หลายครั้ง บทความนี้จะพาคุณเข้าใจเชิงลึกเกี่ยวกับกลไก rate limiting ของ Exchange API ยุคใหม่ พร้อมโค้ด production-ready ที่ผมใช้งานจริงกับ HolySheep AI ซึ่งมี latency เฉลี่ยต่ำกว่า 50ms และราคาประหยัดกว่า 85% เมื่อเทียบกับราคามาตรฐาน

ทำความเข้าใจ Rate Limit Architecture

Exchange API สมัยใหม่ใช้ rate limit แบบหลายมิติ (Multi-dimensional Rate Limiting) ซึ่งแตกต่างจาก model แบบเดิมที่มีแค่ requests per minute

Rate Limit Headers ที่ควรรู้จัก

เมื่อส่ง request ไปยัง Exchange API คุณจะได้รับ headers เหล่านี้กลับมา:

# ตัวอย่าง Response Headers จาก Exchange API
HTTP/2 200
x-ratelimit-limit: 1000
x-ratelimit-remaining: 847
x-ratelimit-reset: 1703123456
x-ratelimit-precision: second
content-type: application/json


เมื่อถูก Rate Limit:

HTTP/2 429
retry-after: 15
x-ratelimit-limit: 1000
x-ratelimit-remaining: 0
x-ratelimit-reset: 1703123456

โครงสร้างโปรเจกต์: Production-Ready Rate Limit Handler

ผมจะสร้างโครงสร้างโปรเจกต์ที่ใช้งานได้จริงใน production พร้อมกับ implementation ของ algorithm หลายแบบ

# โครงสร้างโปรเจกต์
project/
├── requirements.txt
├── config.yaml
├── src/
│   ├── __init__.py
│   ├── client.py           # HolySheep API Client
│   ├── rate_limiter.py     # Rate Limit Implementations
│   ├── retry_strategies.py # Retry with backoff
│   ├── circuit_breaker.py  # Circuit Breaker Pattern
│   └── benchmark.py        # Performance Benchmark
└── tests/
    └── test_rate_limiter.py
# requirements.txt
requests>=2.31.0
httpx>=0.25.0
PyYAML>=6.0
tenacity>=8.2.3
pytest>=7.4.0
pytest-asyncio>=0.21.0
aioresponses>=0.7.6
prometheus-client>=0.19.0
# config.yaml - การตั้งค่าสำหรับ HolySheep API
api:
  base_url: "https://api.holysheep.ai/v1"
  api_key: "YOUR_HOLYSHEEP_API_KEY"
  timeout: 30

rate_limits:
  requests_per_minute: 1000
  tokens_per_minute: 150000
  burst_size: 100

retry:
  max_attempts: 5
  base_delay: 1.0
  max_delay: 60.0
  exponential_base: 2

circuit_breaker:
  failure_threshold: 5
  recovery_timeout: 60
  half_open_max_calls: 3

1. Token Bucket Algorithm Implementation

Token Bucket เป็น algorithm ที่เหมาะกับ use case ที่ต้องการรองรับ burst traffic ได้ เช่น การส่ง request จำนวนมากในช่วงสั้นๆ

# src/rate_limiter.py
import time
import threading
import asyncio
from typing import Optional
from dataclasses import dataclass, field
from collections import deque
import logging

logger = logging.getLogger(__name__)


@dataclass
class TokenBucket:
    """Token Bucket Algorithm - รองรับ burst traffic ได้ดี"""
    capacity: float
    refill_rate: float  # tokens per second
    tokens: float = field(init=False)
    last_refill: float = field(init=False)
    lock: threading.Lock = field(default_factory=threading.Lock)
    
    def __post_init__(self):
        self.tokens = self.capacity
        self.last_refill = time.monotonic()
    
    def _refill(self) -> None:
        """เติม tokens ตามเวลาที่ผ่านไป"""
        now = time.monotonic()
        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 acquire(self, tokens: float = 1.0, blocking: bool = False, timeout: Optional[float] = None) -> bool:
        """
        พยายามเข้าถึง tokens
        
        Args:
            tokens: จำนวน tokens ที่ต้องการ
            blocking: รอจนกว่าจะมี tokens หรือไม่
            timeout: ระยะเวลารอสูงสุด (วินาที)
            
        Returns:
            True ถ้าได้ tokens, False ถ้าไม่มี
        """
        start_time = time.monotonic()
        
        while True:
            with self.lock:
                self._refill()
                
                if self.tokens >= tokens:
                    self.tokens -= tokens
                    return True
                
                if not blocking:
                    return False
                
                # คำนวณเวลารอ
                tokens_needed = tokens - self.tokens
                wait_time = tokens_needed / self.refill_rate
                
                if timeout is not None:
                    elapsed = time.monotonic() - start_time
                    if elapsed + wait_time > timeout:
                        return False
                    wait_time = min(wait_time, timeout - elapsed)
            
            time.sleep(min(wait_time, 0.1))  # ตรวจสอบทุก 100ms


class SlidingWindowRateLimiter:
    """Sliding Window Algorithm - แม่นยำกว่า Fixed Window"""
    
    def __init__(self, max_requests: int, window_size: float):
        self.max_requests = max_requests
        self.window_size = window_size
        self.requests = deque()
        self.lock = threading.Lock()
    
    def _cleanup_old_requests(self, current_time: float) -> None:
        """ลบ requests ที่เก่ากว่า window"""
        cutoff = current_time - self.window_size
        while self.requests and self.requests[0] < cutoff:
            self.requests.popleft()
    
    def acquire(self) -> tuple[bool, float]:
        """
        พยายามเข้าถึง rate limit
        
        Returns:
            (success, retry_after): success ถ้าได้, retry_after คือวินาทีที่ต้องรอ
        """
        current_time = time.monotonic()
        
        with self.lock:
            self._cleanup_old_requests(current_time)
            
            if len(self.requests) < self.max_requests:
                self.requests.append(current_time)
                return True, 0.0
            
            # คำนวณเวลารอ
            oldest = self.requests[0]
            retry_after = (oldest + self.window_size) - current_time
            return False, max(0.0, retry_after)


class AsyncTokenBucket:
    """Async Token Bucket - สำหรับ asyncio applications"""
    
    def __init__(self, capacity: float, refill_rate: float):
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = capacity
        self.last_refill = time.monotonic()
        self._lock = asyncio.Lock()
    
    async def _refill(self) -> None:
        now = time.monotonic()
        elapsed = now - self.last_refill
        new_tokens = elapsed * self.refill_rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_refill = now
    
    async def acquire(self, tokens: float = 1.0) -> None:
        while True:
            async with self._lock:
                await self._refill()
                if self.tokens >= tokens:
                    self.tokens -= tokens
                    return
                tokens_needed = tokens - self.tokens
                wait_time = tokens_needed / self.refill_rate
            
            await asyncio.sleep(min(wait_time, 0.1))



Factory function

def create_rate_limiter(limiter_type: str, **kwargs) -> any:
    """สร้าง rate limiter ตาม type"""
    limiters = {
        'token_bucket': TokenBucket,
        'sliding_window': SlidingWindowRateLimiter,
        'async_token_bucket': AsyncTokenBucket,
    }
    
    if limiter_type not in limiters:
        raise ValueError(f"Unknown limiter type: {limiter_type}")
    
    return limiters[limiter_type](**kwargs)

2. HolySheep API Client with Rate Limiting

นี่คือ production-ready client ที่ผมใช้งานจริงกับ HolySheep AI ซึ่งรองรับ rate limiting แบบหลายมิติ

# src/client.py
import time
import httpx
import asyncio
from typing import Optional, Any, Dict, List
from dataclasses import dataclass
from pathlib import Path
import yaml
import logging

from rate_limiter import TokenBucket, SlidingWindowRateLimiter, AsyncTokenBucket
from retry_strategies import ExponentialBackoff, CircuitBreaker

logger = logging.getLogger(__name__)


@dataclass
class RateLimitConfig:
    """การตั้งค่า Rate Limit สำหรับ Exchange API"""
    requests_per_minute: int = 1000
    tokens_per_minute: int = 150000
    burst_size: int = 100
    cost_per_request: int = 1  # tokens ที่ใช้ต่อ request


class HolySheepClient:
    """
    Production-ready client สำหรับ HolySheep AI API
    รองรับ rate limiting, retry, circuit breaker
    """
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(
        self,
        api_key: str,
        rate_limit_config: Optional[RateLimitConfig] = None,
        max_retries: int = 5,
        timeout: float = 30.0,
        enable_circuit_breaker: bool = True
    ):
        self.api_key = api_key
        self.rate_limit_config = rate_limit_config or RateLimitConfig()
        self.timeout = timeout
        
        # Rate Limiters
        self.request_limiter = TokenBucket(
            capacity=self.rate_limit_config.burst_size,
            refill_rate=self.rate_limit_config.requests_per_minute / 60.0
        )
        self.token_limiter = TokenBucket(
            capacity=self.rate_limit_config.tokens_per_minute,
            refill_rate=self.rate_limit_config.tokens_per_minute / 60.0
        )
        
        # Retry Strategy
        self.retry_strategy = ExponentialBackoff(
            max_attempts=max_retries,
            base_delay=1.0,
            max_delay=60.0,
            exponential_base=2.0
        )
        
        # Circuit Breaker
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60.0,
            half_open_max_calls=3
        ) if enable_circuit_breaker else None
        
        # HTTP Client
        self._client = httpx.Client(
            base_url=self.BASE_URL,
            timeout=timeout,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            }
        )
        
        # Metrics
        self._metrics = {
            "total_requests": 0,
            "successful_requests": 0,
            "rate_limited_requests": 0,
            "failed_requests": 0,
            "total_tokens_used": 0
        }
    
    def _handle_rate_limit(self, response: httpx.Response) -> float:
        """จัดการเมื่อเจอ rate limit"""
        if response.status_code == 429:
            retry_after = float(response.headers.get("Retry-After", 60))
            self._metrics["rate_limited_requests"] += 1
            logger.warning(f"Rate limited. Retry after {retry_after:.2f}s")
            return retry_after
        return 0
    
    def _execute_request(
        self,
        method: str,
        endpoint: str,
        **kwargs
    ) -> httpx.Response:
        """Execute HTTP request โดยตรง"""
        return self._client.request(method, endpoint, **kwargs)
    
    def chat_completions(
        self,
        model: str = "gpt-4.1",
        messages: List[Dict[str, str]],
        temperature: float = 0.7,
        max_tokens: int = 1000,
        **kwargs
    ) -> Dict[str, Any]:
        """
        ส่ง request ไปยัง Chat Completions API
        
        Args:
            model: โมเดลที่ต้องการใช้ (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2)
            messages: รายการ messages
            temperature: ค่า temperature
            max_tokens: จำนวน tokens สูงสุดที่ต้องการ
            
        Returns:
            Response จาก API
        """
        # Acquire rate limit tokens
        self.request_limiter.acquire(blocking=True, timeout=60)
        self.token_limiter.acquire(
            tokens=kwargs.get('estimated_tokens', max_tokens),
            blocking=True,
            timeout=120
        )
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            **kwargs
        }
        
        # Execute with retry
        return self.retry_strategy.execute(
            self._make_request,
            "POST",
            "/chat/completions",
            json=payload
        )
    
    def _make_request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
        """Make request with circuit breaker protection"""
        
        def _do_request():
            response = self._execute_request(method, endpoint, **kwargs)
            
            # Handle rate limit
            retry_after = self._handle_rate_limit(response)
            if retry_after > 0:
                time.sleep(retry_after)
                raise RateLimitError(retry_after)
            
            # Handle other errors
            if response.status_code >= 400:
                raise APIError(
                    f"API request failed: {response.status_code}",
                    status_code=response.status_code,
                    response=response.json() if response.text else None
                )
            
            self._metrics["successful_requests"] += 1
            return response.json()
        
        if self.circuit_breaker:
            return self.circuit_breaker.call(_do_request)
        return _do_request()
    
    def close(self):
        """ปิด HTTP client"""
        self._client.close()
    
    def get_metrics(self) -> Dict[str, Any]:
        """ดึง metrics ปัจจุบัน"""
        return {
            **self._metrics,
            "success_rate": (
                self._metrics["successful_requests"] / 
                max(1, self._metrics["total_requests"]) * 100
            )
        }


class RateLimitError(Exception):
    """Exception เมื่อถูก rate limit"""
    def __init__(self, retry_after: float):
        self.retry_after = retry_after
        super().__init__(f"Rate limited. Retry after {retry_after:.2f}s")


class APIError(Exception):
    """Exception สำหรับ API errors อื่นๆ"""
    def __init__(self, message: str, status_code: int = None, response: dict = None):
        super().__init__(message)
        self.status_code = status_code
        self.response = response



Async Version

class AsyncHolySheepClient:
    """Async version สำหรับ asyncio applications"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str, rate_limit_config: Optional[RateLimitConfig] = None):
        self.api_key = api_key
        self.rate_limit_config = rate_limit_config or RateLimitConfig()
        
        self.request_limiter = AsyncTokenBucket(
            capacity=rate_limit_config.burst_size,
            refill_rate=rate_limit_config.requests_per_minute / 60.0
        )
        self.token_limiter = AsyncTokenBucket(
            capacity=rate_limit_config.tokens_per_minute,
            refill_rate=rate_limit_config.tokens_per_minute / 60.0
        )
        
        self._client = httpx.AsyncClient(
            base_url=self.BASE_URL,
            timeout=30.0,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            }
        )
    
    async def chat_completions(
        self,
        model: str,
        messages: List[Dict[str, str]],
        max_tokens: int = 1000
    ) -> Dict[str, Any]:
        """Async chat completions"""
        await self.request_limiter.acquire()
        await self.token_limiter.acquire(tokens=max_tokens)
        
        response = await self._client.post(
            "/chat/completions",
            json={
                "model": model,
                "messages": messages,
                "max_tokens": max_tokens
            }
        )
        
        if response.status_code == 429:
            retry_after = float(response.headers.get("Retry-After", 60))
            await asyncio.sleep(retry_after)
            return await self.chat_completions(model, messages, max_tokens)
        
        response.raise_for_status()
        return response.json()
    
    async def close(self):
        await self._client.aclose()



Load config from file

def load_config(config_path: str = "config.yaml") -> dict:
    """โหลดการตั้งค่าจากไฟล์"""
    path = Path(config_path)
    if path.exists():
        with open(path) as f:
            return yaml.safe_load(f)
    return {}

3. Retry Strategies with Exponential Backoff

# src/retry_strategies.py
import time
import random
import asyncio
import logging
from typing import Callable, TypeVar, Any
from dataclasses import dataclass
from functools import wraps

logger = logging.getLogger(__name__)

T = TypeVar('T')


class RateLimitAwareRetry:
    """
    Retry strategy ที่รู้จัก Rate Limit headers
    ใช้ Retry-After header แทน fixed delay
    """
    
    def __init__(
        self,
        max_attempts: int = 5,
        base_delay: float = 1.0,
        max_delay: float = 60.0,
        exponential_base: float = 2.0,
        jitter: bool = True
    ):
        self.max_attempts = max_attempts
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.exponential_base = exponential_base
        self.jitter = jitter
    
    def _calculate_delay(self, attempt: int, retry_after: float = None) -> float:
        """คำนวณ delay สำหรับ attempt ปัจจุบัน"""
        # ถ้ามี Retry-After header ใช้ค่านั้นก่อน
        if retry_after:
            return retry_after
        
        # คำนวณ exponential delay
        delay = self.base_delay * (self.exponential_base ** attempt)
        delay = min(delay, self.max_delay)
        
        # เพิ่ม jitter สำหรับ distributed systems
        if self.jitter:
            delay = delay * (0.5 + random.random())
        
        return delay
    
    def execute(self, func: Callable[..., T], *args, **kwargs) -> T:
        """Execute function พร้อม retry logic"""
        last_exception = None
        
        for attempt in range(self.max_attempts):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                last_exception = e
                
                # ดึง Retry-After จาก exception
                retry_after = getattr(e, 'retry_after', None)
                
                if attempt < self.max_attempts - 1:
                    delay = self._calculate_delay(attempt, retry_after)
                    logger.warning(
                        f"Attempt {attempt + 1} failed: {e}. "
                        f"Retrying in {delay:.2f}s..."
                    )
                    time.sleep(delay)
                else:
                    logger.error(f"All {self.max_attempts} attempts failed")
        
        raise last_exception


class CircuitBreaker:
    """
    Circuit Breaker Pattern
    ป้องกันไม่ให้ระบบพยายาม request ไปยัง service ที่กำลังล่ม
    """
    
    class CircuitState:
        CLOSED = "closed"
        OPEN = "open"
        HALF_OPEN = "half_open"
    
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
        half_open_max_calls: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
        
        self.state = self.CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self.half_open_calls = 0
    
    def _should_allow_request(self) -> bool:
        """ตรวจสอบว่าควรอนุญาต request หรือไม่"""
        if self.state == self.CircuitState.CLOSED:
            return True
        
        if self.state == self.CircuitState.OPEN:
            # ตรวจสอบว่าถึงเวลา recovery แล้วหรือยัง
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                self.state = self.CircuitState.HALF_OPEN
                self.half_open_calls = 0
                return True
            return False
        
        # HALF_OPEN state
        if self.half_open_calls < self.half_open_max_calls:
            self.half_open_calls += 1
            return True
        return False
    
    def record_success(self) -> None:
        """บันทึกความสำเร็จ"""
        if self.state == self.CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.half_open_max_calls:
                self.state = self.CircuitState.CLOSED
                self.failure_count = 0
                self.success_count = 0
        else:
            self.failure_count = 0
    
    def record_failure(self) -> None:
        """บันทึกความล้มเหลว"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == self.CircuitState.HALF_OPEN:
            self.state = self.CircuitState.OPEN
            self.success_count = 0
        elif self.failure_count >= self.failure_threshold:
            self.state = self.CircuitState.OPEN
    
    def call(self, func: Callable[..., T], *args, **kwargs) -> T:
        """Execute function พร้อม circuit breaker protection"""
        if not self._should_allow_request():
            raise CircuitBreakerOpenError(
                f"Circuit breaker is OPEN. Retry after "
                f"{self.recovery_timeout - (time.time() - self.last_failure_time):.2f}s"
            )
        
        try:
            result = func(*args, **kwargs)
            self.record_success()
            return result
        except Exception as e:
            self.record_failure()
            raise
    
    def get_state(self) -> dict:
        """ดึงสถานะ circuit breaker"""
        return {
            "state": self.state,
            "failure_count": self.failure_count,
            "success_count": self.success_count,
            "last_failure_time": self.last_failure_time
        }


class CircuitBreakerOpenError(Exception):
    """Exception เมื่อ circuit breaker เปิดอยู่"""
    pass



Async versions

class AsyncCircuitBreaker:
    """Async Circuit Breaker"""
    
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 60.0):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = "closed"
        self.failure_count = 0
        self.last_failure_time = None
        self._lock = asyncio.Lock()
    
    async def call(self, func: Callable[..., T], *args, **kwargs) -> T:
        async with self._lock:
            if self.state == "open":
                elapsed = time.time() - self.last_failure_time
                if elapsed < self.recovery_timeout:
                    raise CircuitBreakerOpenError("Circuit breaker is OPEN")
                self.state = "half_open"
        
        try:
            result = await func(*args, **kwargs)
            async with self._lock:
                self.state = "closed"
                self.failure_count = 0
            return result
        except Exception as e:
            async with self._lock:
                self.failure_count += 1
                self.last_failure_time = time.time()
                if self.failure_count >= self.failure_threshold:
                    self.state = "open"
            raise

4. Performance Benchmark

ผมทดสอบ performance ของ rate limiter implementations ต่างๆ บนเครื่อง MacBook Pro M3 ผลลัพธ์แสดงให้เห็นว่า Token Bucket มี throughput สูงสุดแต่ใช้ CPU มากกว่าเมื่อมี concurrent requests

# src/benchmark.py
import time
import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from typing import List, Callable
import statistics
from dataclasses import dataclass

from rate_limiter import TokenBucket, SlidingWindowRateLimiter, AsyncTokenBucket
from client import HolySheepClient, RateLimitConfig
from retry_strategies import RateLimitAwareRetry, CircuitBreaker


@dataclass
class BenchmarkResult:
    name: str
    total_requests: int
    duration_seconds: float
    requests_per_second: float
    avg_latency_ms: float
    p95_latency_ms: float
    p99_latency_ms: float
    success_rate: float


def benchmark_rate_limiter(
    limiter_type: str,
    limiter: any,
    num_requests: int = 10000,
    concurrency: int = 100,
    tokens_per_request: float = 1.0
) -> BenchmarkResult:
    """Benchmark rate limiter implementations"""
    
    latencies = []
    successes = 0
    failures = 0
    
    def worker(worker_id: int):
        nonlocal successes, failures
        start = time.perf_counter()
        
        if limiter_type.startswith('async'):
            # Async benchmark
            pass
        else:
            acquired = limiter.acquire(tokens_per_request)
            
            if acquired:
                elapsed = time.perf_counter() - start
                latencies.append(elapsed * 1000)  # Convert to ms
                successes += 1
            else:
                failures += 1
    
    start_time = time.perf_counter()
    
    with ThreadPoolExecutor(max_workers=concurrency) as executor:
        futures = [executor.submit(worker, i) for i in range(num_requests)]
        for f in futures:
            f.result()
    
    duration = time.perf_counter() - start_time
    latencies.sort()
    
    return BenchmarkResult(
        name=limiter_type,
        total_requests=num_requests,
        duration_seconds=duration,
        requests_per_second=num_requests / duration,
        avg_latency_ms=statistics.mean(latencies) if latencies else 0,
        p95_latency_ms=latencies[int(len(latencies) * 0.95)] if latencies else 0,
        p99_latency_ms=latencies[int(len(latencies) * 0.99)] if latencies else 0,
        success_rate=successes / num_requests * 100
    )


def benchmark_retry_strategy(
    strategy: RateLimitAwareRetry,
    func: Callable,
    error_rate: float = 0.3,
    num_trials: int = 1000
) -> dict:
    """Benchmark retry strategy"""
    
    results = {
        "total_trials": num_trials,
        "successful_trials": 0,
        "total_attempts": 0,
        "avg_attempts": 0,
        "total_time": 0,
        "rate_limited_calls": 0
    }
    
    call_count = [0]
    
    def flaky_function():
        call_count[0] += 1
        results["total_attempts"] += 1
        
        # Simulate rate limit occasionally
        if call_count[0] % 10 == 0:
            results["rate_limited_calls"] += 1
            error = Exception("429 Rate Limited")
            error.retry_after = 0.1
            raise error
        
        if random.random() < error_rate:
            raise Exception("Random error")
        
        results["successful_trials"] += 1
        return "success"
    
    start = time.perf_counter()
    
    for _ in range(num_trials):
        try:
            strategy.execute(flaky_function)
        except Exception:
            pass
    
    results["total_time"] = time.perf_counter() - start
    results["avg_attempts"] = results["total_attempts"] / num_trials
    
    return results


async def benchmark_async_client():
    """Benchmark async

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