03-1. 案例：I/O 密集型應用

# I/O 密集型應用的特徵

✅ 大量外部調用：
   - 數據庫查詢（PostgreSQL、MySQL、MongoDB）
   - 緩存讀寫（Redis、Memcached）
   - 外部 API 調用（支付、簡訊、天氣 API）
   - 文件操作（讀寫、上傳、下載）
   - 消息隊列（RabbitMQ、Kafka）

✅ 響應時間特徵：
   - 平均響應時間：100-500ms
   - P95 響應時間：500ms-2s
   - 時間主要花在等待而非計算

❌ 不是 I/O 密集型：
   - 影像處理（resize、濾鏡）
   - 視頻轉碼
   - 大量數學計算
   - 機器學習推理
   - 加密解密（大量數據）

# myapp/views.py
import httpx
from django.http import JsonResponse

async def api_gateway_view(request):
    """
    API 網關：需要調用多個外部服務
    這是典型的 I/O 密集型場景
    """
    async with httpx.AsyncClient() as client:
        # 並發調用多個服務
        user_task = client.get('http://user-service/api/user/123')
        order_task = client.get('http://order-service/api/orders?user=123')
        product_task = client.get('http://product-service/api/products')

        # 等待所有請求完成
        user_resp, order_resp, product_resp = await asyncio.gather(
            user_task,
            order_task,
            product_task
        )

    return JsonResponse({
        'user': user_resp.json(),
        'orders': order_resp.json(),
        'products': product_resp.json(),
    })

# myapp/views.py
from django.db import models
from django.http import JsonResponse

class Order(models.Model):
    user = models.ForeignKey('User', on_delete=models.CASCADE)
    product = models.ForeignKey('Product', on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)

async def order_list_view(request):
    """
    數據庫密集型：大量查詢和關聯
    """
    # 異步 ORM 查詢（Django 4.1+）
    orders = await Order.objects.select_related(
        'user', 'product'
    ).prefetch_related(
        'items', 'payments'
    ).filter(
        created_at__gte=timezone.now() - timedelta(days=30)
    ).aall()  # Django 4.1+ 異步查詢

    return JsonResponse({
        'orders': [
            {
                'id': order.id,
                'user': order.user.name,
                'product': order.product.name,
                'total': order.total,
            }
            for order in orders
        ]
    })

# myapp/services.py
import httpx
from django.core.cache import cache

class UserService:
    """用戶微服務客戶端"""

    async def get_user(self, user_id):
        # 先查緩存
        cache_key = f'user:{user_id}'
        user = cache.get(cache_key)
        if user:
            return user

        # 緩存未命中，調用 API
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f'http://user-service/users/{user_id}'
            )
            user = resp.json()

        # 寫入緩存
        cache.set(cache_key, user, timeout=300)
        return user

async def dashboard_view(request):
    """儀表板：整合多個微服務數據"""
    service = UserService()

    # 並發調用
    user_data = await service.get_user(request.user.id)
    stats_data = await fetch_user_stats(request.user.id)
    recent_activity = await fetch_recent_activity(request.user.id)

    return render(request, 'dashboard.html', {
        'user': user_data,
        'stats': stats_data,
        'activity': recent_activity,
    })

# gunicorn.conf.py - I/O 密集型應用專用配置
import multiprocessing
import os

# ===== Worker 配置 =====
# I/O 密集型：使用 ASGI 異步 Worker
bind = "0.0.0.0:8000"
workers = multiprocessing.cpu_count()  # CPU 核心數即可
worker_class = "uvicorn.workers.UvicornWorker"  # 異步 Worker
worker_connections = 1000  # 每個 worker 的並發連接數

# ===== 超時配置 =====
# I/O 密集型應用通常需要較長的超時時間
timeout = 120  # 2 分鐘（考慮外部 API 可能很慢）
graceful_timeout = 30
keepalive = 20  # 較長的 keepalive 適合持久連接

# ===== 防止記憶體洩漏 =====
max_requests = 2000
max_requests_jitter = 200

# ===== 性能優化 =====
preload_app = True
worker_tmp_dir = "/dev/shm"

# ===== 日誌配置 =====
loglevel = "info"
accesslog = "/var/log/gunicorn/access.log"
errorlog = "/var/log/gunicorn/error.log"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'

# ===== Hooks =====
def post_fork(server, worker):
    """Worker 初始化時關閉資料庫連接"""
    from django import db
    db.connections.close_all()
    server.log.info(f"Worker {worker.pid} spawned (I/O intensive mode)")

def worker_exit(server, worker):
    from django import db
    db.connections.close_all()
    server.log.info(f"Worker {worker.pid} exited")

# 1. Workers 數量：CPU 核心數
workers = multiprocessing.cpu_count()

# 原因：
# - ASGI 異步模型，一個 worker 可處理數千並發
# - I/O 等待時，worker 不會閒置，會處理其他請求
# - 不需要像 Sync Worker 那樣 (2 × CPU + 1)

# 例如 4 核 CPU：
# - Sync Worker 需要：(2 × 4) + 1 = 9 個
# - Async Worker 只需：4 個
# - 但並發處理能力更強！

# 2. Worker 類型：Uvicorn
worker_class = "uvicorn.workers.UvicornWorker"

# 原因：
# - 原生支援 async/await
# - 高效的事件循環（asyncio）
# - 一個 worker 可同時處理數千個 I/O 等待的請求

# 3. 較長的 timeout
timeout = 120

# 原因：
# - 外部 API 可能響應慢
# - 數據庫複雜查詢可能需要時間
# - 防止正常的慢請求被誤殺

# 4. 較長的 keepalive
keepalive = 20

# 原因：
# - I/O 密集型應用適合持久連接
# - 減少 TCP 握手開銷
# - 提高性能

# test_app/views.py
import asyncio
import httpx
from django.http import JsonResponse

# 模擬 I/O 密集型視圖
async def io_intensive_view(request):
    """模擬調用 3 個外部服務，每個需要 300ms"""
    async with httpx.AsyncClient() as client:
        tasks = [
            client.get('http://httpbin.org/delay/0.3'),
            client.get('http://httpbin.org/delay/0.3'),
            client.get('http://httpbin.org/delay/0.3'),
        ]
        results = await asyncio.gather(*tasks)

    return JsonResponse({'status': 'ok', 'count': len(results)})

# 同步版本（用於對比）
def sync_io_intensive_view(request):
    """同步版本：按順序調用"""
    import requests
    results = []
    for i in range(3):
        resp = requests.get('http://httpbin.org/delay/0.3')
        results.append(resp.json())

    return JsonResponse({'status': 'ok', 'count': len(results)})

# 測試命令
ab -n 1000 -c 100 http://localhost:8000/api/test/

# gunicorn.conf.py
workers = 9  # (2 × 4) + 1
worker_class = "sync"

# gunicorn.conf.py
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"

# gunicorn.conf.py
workers = 9
worker_class = "gevent"
worker_connections = 1000

# ecommerce/views.py
import asyncio
import httpx
from django.http import JsonResponse
from django.core.cache import cache
from django.db import models

class Product(models.Model):
    name = models.CharField(max_length=200)
    description = models.TextField()
    price = models.DecimalField(max_digits=10, decimal_places=2)

async def product_detail_view(request, product_id):
    """商品詳情 API - I/O 密集型"""

    # 並發執行多個 I/O 操作
    product_task = get_product_info(product_id)
    stock_task = get_stock_from_redis(product_id)
    price_task = get_price_from_service(product_id)
    recommend_task = get_recommendations(product_id)
    favorite_task = check_user_favorite(request.user.id, product_id)

    # 等待所有任務完成
    product, stock, price, recommendations, is_favorite = await asyncio.gather(
        product_task,
        stock_task,
        price_task,
        recommend_task,
        favorite_task,
        return_exceptions=True  # 某個服務失敗不影響其他
    )

    return JsonResponse({
        'product': {
            'id': product.id,
            'name': product.name,
            'description': product.description,
        },
        'stock': stock if not isinstance(stock, Exception) else 0,
        'price': price if not isinstance(price, Exception) else product.price,
        'recommendations': recommendations if not isinstance(recommendations, Exception) else [],
        'is_favorite': is_favorite if not isinstance(is_favorite, Exception) else False,
    })


async def get_product_info(product_id):
    """查詢商品資訊（約 80ms）"""
    # Django 4.1+ 異步 ORM
    return await Product.objects.aget(id=product_id)


async def get_stock_from_redis(product_id):
    """從 Redis 查詢庫存（約 10ms）"""
    from django.core.cache import cache
    stock = cache.get(f'stock:{product_id}')
    return stock if stock is not None else 0


async def get_price_from_service(product_id):
    """調用外部價格服務（約 150ms）"""
    async with httpx.AsyncClient(timeout=2.0) as client:
        try:
            resp = await client.get(
                f'http://price-service/api/price/{product_id}'
            )
            return resp.json()['price']
        except:
            return None


async def get_recommendations(product_id):
    """調用推薦引擎（約 200ms）"""
    async with httpx.AsyncClient(timeout=3.0) as client:
        try:
            resp = await client.get(
                f'http://recommend-service/api/recommend/{product_id}'
            )
            return resp.json()['products']
        except:
            return []


async def check_user_favorite(user_id, product_id):
    """查詢用戶收藏（約 50ms）"""
    # 假設使用 MongoDB
    from motor.motor_asyncio import AsyncIOMotorClient
    client = AsyncIOMotorClient('mongodb://localhost:27017')
    db = client.ecommerce

    favorite = await db.favorites.find_one({
        'user_id': user_id,
        'product_id': product_id
    })

    return favorite is not None

# gunicorn.conf.py
import multiprocessing

# 伺服器規格：8 核 CPU，16GB RAM
bind = "0.0.0.0:8000"
workers = 8  # = CPU 核心數
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000

# 超時配置
timeout = 120
graceful_timeout = 30
keepalive = 20

# 防止記憶體洩漏
max_requests = 3000
max_requests_jitter = 300

# 性能優化
preload_app = True
worker_tmp_dir = "/dev/shm"

# 日誌
loglevel = "warning"
accesslog = "/var/log/gunicorn/ecommerce-access.log"
errorlog = "/var/log/gunicorn/ecommerce-error.log"

# Hooks
def post_fork(server, worker):
    from django import db
    from django.core.cache import cache

    # 關閉資料庫連接
    db.connections.close_all()

    # 測試 Redis 連接
    try:
        cache.get('health_check')
    except Exception as e:
        server.log.error(f"Redis connection failed: {e}")

    server.log.info(f"Worker {worker.pid} ready for I/O intensive workload")

# 使用 wrk 進行壓測
wrk -t 8 -c 1000 -d 60s http://localhost:8000/api/products/123/

# 結果：
Running 60s test @ http://localhost:8000/api/products/123/
  8 threads and 1000 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   185.32ms   45.23ms   1.50s    89.45%
    Req/Sec     687.23    156.78     1.20k    78.23%

  328,754 requests in 60.00s, 156.32MB read

Requests/sec:   5479.23  ← 超過目標 1000 QPS ✅
Transfer/sec:      2.61MB

# 方案 1：增加 Gunicorn timeout
timeout = 180  # 從 120 增加到 180

# 方案 2：為外部請求設定獨立的 timeout（推薦）
async with httpx.AsyncClient(timeout=10.0) as client:  # 10 秒超時
    try:
        resp = await client.get(external_url)
    except httpx.TimeoutException:
        # 優雅處理超時
        return default_value

# 方案 3：使用降級策略
async def fetch_with_fallback(url, fallback_value):
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            resp = await client.get(url)
            return resp.json()
    except Exception as e:
        logger.warning(f"API call failed: {e}, using fallback")
        return fallback_value

# settings.py

# 方案 1：配置合理的連接池大小
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'mydb',
        'CONN_MAX_AGE': 600,  # 持久連接
        # 計算：200 max_connections / 8 workers = 25 per worker
        'OPTIONS': {
            'connect_timeout': 10,
        }
    }
}

# 方案 2：使用連接池中間件（推薦）
# pip install django-db-connection-pool

DATABASES = {
    'default': {
        'ENGINE': 'dj_db_conn_pool.backends.postgresql',
        'POOL_OPTIONS': {
            'POOL_SIZE': 10,  # 每個 worker 最多 10 個連接
            'MAX_OVERFLOW': 5,
        }
    }
}

# 總連接數：8 workers × 10 = 80（安全範圍內）

# 方案 3：在 post_fork Hook 中配置
def post_fork(server, worker):
    from django import db
    from django.db.backends.postgresql.base import DatabaseWrapper

    # 關閉繼承的連接
    db.connections.close_all()

    # 為每個 worker 設定連接限制
    for conn in db.connections.all():
        conn.settings_dict['CONN_MAX_AGE'] = 300

# settings.py

# 使用連接池
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'OPTIONS': {
            'CLIENT_CLASS': 'django_redis.client.DefaultClient',
            'CONNECTION_POOL_KWARGS': {
                'max_connections': 50,  # 每個進程最多 50 連接
                'retry_on_timeout': True,
            },
            'SOCKET_CONNECT_TIMEOUT': 5,
            'SOCKET_TIMEOUT': 5,
        }
    }
}

# 計算：8 workers × 50 = 400 connections
# 確保 Redis maxclients 設定夠大（預設 10000）

# 記憶體使用不斷增長
ps aux | grep gunicorn
# Worker 記憶體從 200MB 增長到 1GB+

# gunicorn.conf.py

# 方案 1：定期重啟 workers
max_requests = 2000  # 處理 2000 請求後重啟
max_requests_jitter = 200  # 加入隨機性避免同時重啟

# 方案 2：記憶體限制重啟
import psutil

def pre_request(worker, req):
    """每個請求前檢查記憶體"""
    process = psutil.Process(worker.pid)
    mem_mb = process.memory_info().rss / 1024 / 1024

    # 如果 worker 記憶體超過 500MB，標記重啟
    if mem_mb > 500:
        worker.log.warning(f"Worker {worker.pid} memory: {mem_mb}MB, will restart soon")
        # 可以設定 worker.alive = False 強制重啟

# 方案 3：使用垃圾回收
def post_request(worker, req, environ, resp):
    """每 100 個請求手動觸發 GC"""
    import gc
    if not hasattr(worker, 'request_count'):
        worker.request_count = 0

    worker.request_count += 1
    if worker.request_count % 100 == 0:
        gc.collect()

# 推薦配置（Django 3.0+）
workers = multiprocessing.cpu_count()
worker_class = "uvicorn.workers.UvicornWorker"
timeout = 120
keepalive = 20