# 

# 05-6. Server-Sent Events (SSE)

> ⏱️ **閱讀時間：** 15 分鐘
> 🎯 **難度：** ⭐⭐⭐ (進階)

---

## 🤔 一句話解釋

**Server-Sent Events (SSE) 是伺服器單向推送資料給客戶端的技術，比 WebSocket 更簡單，適合即時通知場景。**

---

## 🔄 WebSocket vs SSE

```
WebSocket（雙向通訊）：
┌────────┐ ◀═══════▶ ┌────────┐
│ Client │   雙向    │ Server │
└────────┘           └────────┘
客戶端和伺服器都可以主動發送

SSE（單向推送）：
┌────────┐ ◀──────── ┌────────┐
│ Client │   單向    │ Server │
└────────┘           └────────┘
只有伺服器可以推送
```

### 比較表

| 特性 | WebSocket | SSE |
|------|-----------|-----|
| **方向** | 雙向 | 單向（伺服器 → 客戶端）|
| **協議** | ws:// | http:// |
| **複雜度** | 較高 | 較低 |
| **自動重連** | 需自行實作 | 瀏覽器內建 |
| **二進制** | 支援 | 不支援（純文字）|
| **適用場景** | 聊天、遊戲 | 通知、即時更新 |

---

## 🎯 適用場景

| 場景 | 說明 |
|------|------|
| **即時通知** | 新訊息、系統警告 |
| **股票報價** | 價格即時更新 |
| **進度更新** | 檔案上傳、任務進度 |
| **社群動態** | 新貼文、新留言 |
| **日誌串流** | 即時日誌輸出 |

---

## 📦 FastAPI SSE 基礎

### 基本範例

```python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def event_generator():
    """產生 SSE 事件"""
    count = 0
    while True:
        count += 1
        # SSE 格式：data: 內容\n\n
        yield f"data: 訊息 {count}\n\n"
        await asyncio.sleep(1)

@app.get("/events")
async def sse_endpoint():
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        }
    )
```

### 客戶端接收

```javascript
// 瀏覽器 JavaScript
const eventSource = new EventSource("/events");

eventSource.onmessage = (event) => {
    console.log("收到:", event.data);
};

eventSource.onerror = (error) => {
    console.error("連線錯誤:", error);
};

// 關閉連線
// eventSource.close();
```

---

## 🔧 SSE 訊息格式

### 標準格式

```python
async def event_generator():
    # 基本訊息
    yield "data: Hello\n\n"

    # 多行訊息
    yield "data: 第一行\n"
    yield "data: 第二行\n\n"

    # 指定事件類型
    yield "event: notification\n"
    yield "data: 新通知\n\n"

    # 指定 ID（用於重連）
    yield "id: 123\n"
    yield "data: 訊息內容\n\n"

    # 設定重試間隔（毫秒）
    yield "retry: 5000\n"
    yield "data: 訊息\n\n"
```

### 客戶端處理不同事件

```javascript
const eventSource = new EventSource("/events");

// 預設訊息（沒有 event 欄位）
eventSource.onmessage = (event) => {
    console.log("預設:", event.data);
};

// 自訂事件類型
eventSource.addEventListener("notification", (event) => {
    console.log("通知:", event.data);
});

eventSource.addEventListener("update", (event) => {
    console.log("更新:", event.data);
});
```

---

## 📊 實用範例

### JSON 資料推送

```python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from datetime import datetime
import asyncio
import json

app = FastAPI()

class StockPrice(BaseModel):
    symbol: str
    price: float
    timestamp: str

async def stock_generator(symbol: str):
    """模擬股票價格推送"""
    import random

    base_price = 100.0

    while True:
        # 模擬價格波動
        change = random.uniform(-1, 1)
        base_price += change

        stock = StockPrice(
            symbol=symbol,
            price=round(base_price, 2),
            timestamp=datetime.utcnow().isoformat()
        )

        yield f"data: {stock.model_dump_json()}\n\n"
        await asyncio.sleep(1)

@app.get("/stocks/{symbol}")
async def stock_stream(symbol: str):
    return StreamingResponse(
        stock_generator(symbol),
        media_type="text/event-stream"
    )
```

### 任務進度更新

```python
from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import StreamingResponse
import asyncio
from typing import Dict
import uuid

app = FastAPI()

# 儲存任務狀態
task_progress: Dict[str, dict] = {}

async def long_running_task(task_id: str):
    """模擬長時間任務"""
    for i in range(1, 101):
        task_progress[task_id] = {
            "progress": i,
            "status": "processing" if i < 100 else "completed"
        }
        await asyncio.sleep(0.1)

@app.post("/tasks")
async def create_task(background_tasks: BackgroundTasks):
    """建立新任務"""
    task_id = str(uuid.uuid4())
    task_progress[task_id] = {"progress": 0, "status": "pending"}

    background_tasks.add_task(long_running_task, task_id)

    return {"task_id": task_id}

async def progress_generator(task_id: str):
    """產生進度事件"""
    last_progress = -1

    while True:
        if task_id not in task_progress:
            yield f"data: {json.dumps({'error': 'Task not found'})}\n\n"
            break

        current = task_progress[task_id]

        # 只在進度改變時推送
        if current["progress"] != last_progress:
            yield f"data: {json.dumps(current)}\n\n"
            last_progress = current["progress"]

        if current["status"] == "completed":
            break

        await asyncio.sleep(0.1)

@app.get("/tasks/{task_id}/progress")
async def task_progress_stream(task_id: str):
    return StreamingResponse(
        progress_generator(task_id),
        media_type="text/event-stream"
    )
```

### 即時日誌串流

```python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from collections import deque
import asyncio
import json
from datetime import datetime

app = FastAPI()

# 日誌緩衝區
log_buffer: deque = deque(maxlen=1000)
log_subscribers: list = []

class LogEntry:
    def __init__(self, level: str, message: str):
        self.level = level
        self.message = message
        self.timestamp = datetime.utcnow().isoformat()

    def to_dict(self):
        return {
            "level": self.level,
            "message": self.message,
            "timestamp": self.timestamp
        }

def add_log(level: str, message: str):
    """添加日誌"""
    entry = LogEntry(level, message)
    log_buffer.append(entry)

    # 通知所有訂閱者
    for queue in log_subscribers:
        queue.put_nowait(entry)

async def log_generator():
    """產生日誌事件"""
    queue = asyncio.Queue()
    log_subscribers.append(queue)

    try:
        # 先發送歷史日誌
        for entry in list(log_buffer):
            yield f"data: {json.dumps(entry.to_dict())}\n\n"

        # 等待新日誌
        while True:
            entry = await queue.get()
            yield f"data: {json.dumps(entry.to_dict())}\n\n"
    finally:
        log_subscribers.remove(queue)

@app.get("/logs")
async def log_stream():
    return StreamingResponse(
        log_generator(),
        media_type="text/event-stream"
    )

@app.post("/logs")
async def create_log(level: str, message: str):
    add_log(level, message)
    return {"status": "ok"}
```

---

## 🔐 認證

### Token 認證

```python
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import StreamingResponse

app = FastAPI()

def verify_token(token: str) -> bool:
    """驗證 token"""
    # 實際應用中要驗證 JWT 或 session
    return token == "valid_token"

async def authenticated_generator(user_id: str):
    """認證後的事件產生器"""
    count = 0
    while True:
        count += 1
        yield f"data: User {user_id} 的訊息 {count}\n\n"
        await asyncio.sleep(1)

@app.get("/secure-events")
async def secure_sse(token: str = Query(...)):
    if not verify_token(token):
        raise HTTPException(status_code=401, detail="Invalid token")

    return StreamingResponse(
        authenticated_generator("user123"),
        media_type="text/event-stream"
    )
```

### 客戶端帶 Token

```javascript
// 方法一：Query Parameter
const eventSource = new EventSource("/secure-events?token=valid_token");

// 方法二：使用 fetch + ReadableStream
async function connectSSE() {
    const response = await fetch("/secure-events", {
        headers: {
            "Authorization": "Bearer your_token"
        }
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder();

    while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        const text = decoder.decode(value);
        console.log("收到:", text);
    }
}
```

---

## 💓 心跳機制

```python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def heartbeat_generator():
    """帶心跳的事件產生器"""
    heartbeat_interval = 15  # 秒
    last_event_time = asyncio.get_event_loop().time()

    while True:
        current_time = asyncio.get_event_loop().time()

        # 檢查是否需要發送心跳
        if current_time - last_event_time >= heartbeat_interval:
            yield ": heartbeat\n\n"  # 註解格式的心跳
            last_event_time = current_time

        # 檢查是否有實際事件
        event = await check_for_events()
        if event:
            yield f"data: {event}\n\n"
            last_event_time = current_time

        await asyncio.sleep(1)

async def check_for_events():
    """檢查是否有新事件"""
    # 實際應用中從 Redis/資料庫檢查
    import random
    if random.random() < 0.1:
        return "新事件"
    return None

@app.get("/events")
async def sse_with_heartbeat():
    return StreamingResponse(
        heartbeat_generator(),
        media_type="text/event-stream"
    )
```

---

## 📝 完整通知系統

```python
from fastapi import FastAPI, Query
from fastapi.responses import StreamingResponse, HTMLResponse
from pydantic import BaseModel
from typing import Dict, List
import asyncio
import json
from datetime import datetime
import uuid

app = FastAPI()

class Notification(BaseModel):
    id: str
    type: str
    title: str
    message: str
    timestamp: str

class NotificationManager:
    def __init__(self):
        self.subscribers: Dict[str, asyncio.Queue] = {}

    async def subscribe(self, user_id: str):
        """訂閱通知"""
        queue = asyncio.Queue()
        self.subscribers[user_id] = queue
        return queue

    def unsubscribe(self, user_id: str):
        """取消訂閱"""
        self.subscribers.pop(user_id, None)

    async def notify(self, user_id: str, notification: Notification):
        """發送通知給特定用戶"""
        if user_id in self.subscribers:
            await self.subscribers[user_id].put(notification)

    async def broadcast(self, notification: Notification):
        """廣播通知給所有用戶"""
        for queue in self.subscribers.values():
            await queue.put(notification)

manager = NotificationManager()

async def notification_generator(user_id: str):
    """產生通知事件"""
    queue = await manager.subscribe(user_id)

    try:
        while True:
            notification = await queue.get()
            yield f"event: notification\n"
            yield f"id: {notification.id}\n"
            yield f"data: {notification.model_dump_json()}\n\n"
    finally:
        manager.unsubscribe(user_id)

@app.get("/notifications/{user_id}")
async def notification_stream(user_id: str):
    return StreamingResponse(
        notification_generator(user_id),
        media_type="text/event-stream"
    )

@app.post("/notifications/{user_id}")
async def send_notification(user_id: str, title: str, message: str):
    """發送通知給特定用戶"""
    notification = Notification(
        id=str(uuid.uuid4()),
        type="info",
        title=title,
        message=message,
        timestamp=datetime.utcnow().isoformat()
    )
    await manager.notify(user_id, notification)
    return {"status": "sent"}

@app.post("/notifications/broadcast")
async def broadcast_notification(title: str, message: str):
    """廣播通知"""
    notification = Notification(
        id=str(uuid.uuid4()),
        type="broadcast",
        title=title,
        message=message,
        timestamp=datetime.utcnow().isoformat()
    )
    await manager.broadcast(notification)
    return {"status": "broadcasted"}

# HTML 前端
@app.get("/")
async def get_page():
    return HTMLResponse("""
    <!DOCTYPE html>
    <html>
    <head>
        <title>SSE 通知</title>
        <style>
            .notification {
                padding: 10px;
                margin: 5px;
                border: 1px solid #ccc;
                border-radius: 5px;
            }
            .info { background-color: #e3f2fd; }
            .broadcast { background-color: #fff3e0; }
        </style>
    </head>
    <body>
        <h1>SSE 通知系統</h1>
        <div id="notifications"></div>

        <script>
            const userId = "user_" + Math.random().toString(36).substr(2, 9);
            const eventSource = new EventSource(`/notifications/${userId}`);

            eventSource.addEventListener("notification", (event) => {
                const data = JSON.parse(event.data);
                const div = document.createElement("div");
                div.className = `notification ${data.type}`;
                div.innerHTML = `
                    <strong>${data.title}</strong>
                    <p>${data.message}</p>
                    <small>${data.timestamp}</small>
                `;
                document.getElementById("notifications").prepend(div);
            });

            eventSource.onerror = () => {
                console.log("連線中斷，嘗試重連...");
            };
        </script>
    </body>
    </html>
    """)
```

---

## ✅ 重點總結

### SSE 格式

| 欄位 | 說明 |
|------|------|
| `data:` | 訊息內容 |
| `event:` | 事件類型 |
| `id:` | 訊息 ID |
| `retry:` | 重連間隔（毫秒）|

### 最佳實踐

1. **設定適當的 Headers**
   - `Cache-Control: no-cache`
   - `Connection: keep-alive`
2. **實作心跳機制**：防止連線超時
3. **使用 ID**：支援斷線重連
4. **錯誤處理**：客戶端處理 onerror

---

## 🎤 面試這樣答

### Q: SSE 和 WebSocket 的差別？何時選擇 SSE？

**答案：**

> **差別：**
> - SSE 是單向的（伺服器 → 客戶端）
> - WebSocket 是雙向的
>
> **選擇 SSE 的情況：**
> 1. 只需要伺服器推送（通知、更新）
> 2. 不需要客戶端發送大量訊息
> 3. 希望簡化實作（SSE 更簡單）
> 4. 需要自動重連（瀏覽器內建）
>
> ```python
> @app.get("/events")
> async def sse_endpoint():
>     async def generator():
>         while True:
>             yield f"data: {datetime.now()}\n\n"
>             await asyncio.sleep(1)
>
>     return StreamingResponse(
>         generator(),
>         media_type="text/event-stream"
>     )
> ```

---

**上一篇：** [05-5. WebSocket 基礎](./05-5)
**下一篇：** [05-7. 背景任務](./05-7)

---

最後更新：2025-12-18


---

> 作者: luk  
> URL: https://yoru-karu-blog-lalaluk-52581ac5e0cef170a3c8922c19182ecb6f7bd604.gitlab.io/posts/tutorial/fastapi/05-6/