วิธีสตรีม API ด้วย Server-Sent Events (SSE)

สรุป (TL;DR)

ใช้ Server-Sent Events (SSE) เพื่อสตรีมการตอบสนอง API ผ่าน HTTP ส่ง Content-Type: text/event-stream และเขียนเหตุการณ์ในรูปแบบ data: {json}\n\n SSE ใช้ได้กับการสตรีมการตอบกลับของ AI, การอัปเดตความคืบหน้า และฟีดสด Modern PetstoreAPI ใช้ SSE สำหรับคำแนะนำสัตว์เลี้ยงโดย AI และการอัปเดตสถานะคำสั่งซื้อ

บทนำ

API ของคุณสร้างคำแนะนำสัตว์เลี้ยงด้วย AI การตอบสนองใช้เวลา 10 วินาที คุณจะให้ผู้ใช้รอ หรือสตรีมผลลัพธ์ทันทีที่สร้างเสร็จ?

ด้วย Server-Sent Events (SSE) คุณสามารถสตรีมการตอบสนองแบบเรียลไทม์ ผู้ใช้จะเห็นผลลัพธ์ทันทีที่ AI สร้างเสร็จ ทำให้ได้รับประสบการณ์ที่ดีขึ้น

Modern PetstoreAPI ใช้ SSE สำหรับคำแนะนำสัตว์เลี้ยงโดย AI, การอัปเดตสถานะคำสั่งซื้อ และการเปลี่ยนแปลงสินค้าคงคลัง

หากคุณกำลังทดสอบ API สำหรับการสตรีม Apidog รองรับการทดสอบและตรวจสอบ SSE

button

พื้นฐานของ SSE

SSE เป็นการสตรีมแบบทางเดียวจากเซิร์ฟเวอร์ไปยังไคลเอนต์โดยใช้ HTTP

รูปแบบ SSE

Content-Type: text/event-stream
Cache-Control: no-cache

data: {"message":"First chunk"}\n\n
data: {"message":"Second chunk"}\n\n
data: {"message":"Third chunk"}\n\n

แต่ละเหตุการณ์:

เริ่มต้นด้วย data:
มีเพย์โหลด (payload)
ลงท้ายด้วย \n\n (การขึ้นบรรทัดใหม่สองครั้ง)

เหตุการณ์ที่มีชื่อ

event: recommendation
data: {"petId":"019b4132","score":0.95}

event: recommendation
data: {"petId":"019b4127","score":0.89}

event: complete
data: {"total":2}

Event IDs

id: 1
data: {"message":"First"}

id: 2
data: {"message":"Second"}

ไคลเอนต์สามารถดำเนินการต่อจาก ID ล่าสุดได้หากการเชื่อมต่อหลุด

การนำ SSE Server ไปใช้งาน

ตัวอย่าง Node.js/Express

app.get('/v1/pets/recommendations/stream', async (req, res) => {
  // Set SSE headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  // Send recommendations as they're generated
  const recommendations = await generateRecommendations(req.query.userId);

  for (const rec of recommendations) {
    res.write(`data: ${JSON.stringify(rec)}\n\n`);
    await sleep(100); // Simulate streaming delay
  }

  // Send completion event
  res.write(`event: complete\ndata: {"total":${recommendations.length}}\n\n`);
  res.end();
});

ตัวอย่าง Python/FastAPI

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json
import asyncio

app = FastAPI()

@app.get("/v1/pets/recommendations/stream")
async def stream_recommendations(user_id: str):
    async def generate():
        recommendations = await get_recommendations(user_id)

        for rec in recommendations:
            yield f"data: {json.dumps(rec)}\n\n"
            await asyncio.sleep(0.1)

        yield f"event: complete\ndata: {json.dumps({'total': len(recommendations)})}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive"
        }
    )

การนำ SSE Client ไปใช้งาน

JavaScript/Browser

const eventSource = new EventSource(
  'https://petstoreapi.com/v1/pets/recommendations/stream?userId=user-456'
);

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  displayRecommendation(data);
};

eventSource.addEventListener('complete', (event) => {
  const data = JSON.parse(event.data);
  console.log(`Received ${data.total} recommendations`);
  eventSource.close();
});

eventSource.onerror = (error) => {
  console.error('SSE error:', error);
  eventSource.close();
};

React Hook

import { useEffect, useState } from 'react';

function useSSE(url) {
  const [data, setData] = useState([]);
  const [complete, setComplete] = useState(false);

  useEffect(() => {
    const eventSource = new EventSource(url);

    eventSource.onmessage = (event) => {
      const item = JSON.parse(event.data);
      setData(prev => [...prev, item]);
    };

    eventSource.addEventListener('complete', () => {
      setComplete(true);
      eventSource.close();
    });

    return () => eventSource.close();
  }, [url]);

  return { data, complete };
}

// Usage
function Recommendations({ userId }) {
  const { data, complete } = useSSE(
    `https://petstoreapi.com/v1/pets/recommendations/stream?userId=${userId}`
  );

  return (
    <div>
      {data.map(rec => (
        <PetCard key={rec.petId} pet={rec} />
      ))}
      {!complete && <Spinner />}
    </div>
  );
}

Modern PetstoreAPI ใช้ SSE อย่างไร

คำแนะนำสัตว์เลี้ยงด้วย AI

สตรีมคำแนะนำที่สร้างโดย AI:

GET /v1/pets/recommendations/stream?userId=user-456
Accept: text/event-stream

event: recommendation
data: {"petId":"019b4132","name":"Fluffy","score":0.95,"reason":"Matches your preference for cats"}

event: recommendation
data: {"petId":"019b4127","name":"Buddy","score":0.89,"reason":"Similar to pets you liked"}

event: complete
data: {"total":2,"processingTime":850}

การอัปเดตสถานะคำสั่งซื้อ

สตรีมขั้นตอนการประมวลผลคำสั่งซื้อ:

GET /v1/orders/019b4132/status/stream
Accept: text/event-stream

data: {"status":"payment_processing","timestamp":"2026-03-13T10:30:00Z"}

data: {"status":"payment_confirmed","timestamp":"2026-03-13T10:30:02Z"}

data: {"status":"preparing_shipment","timestamp":"2026-03-13T10:30:05Z"}

event: complete
data: {"status":"shipped","trackingNumber":"1Z999AA10123456784"}

การเปลี่ยนแปลงสินค้าคงคลัง

สตรีมการอัปเดตสินค้าคงคลังแบบเรียลไทม์:

GET /v1/inventory/stream
Accept: text/event-stream

event: stock-change
data: {"petId":"019b4132","oldStock":5,"newStock":4}

event: price-change
data: {"petId":"019b4127","oldPrice":299.99,"newPrice":279.99}

ดู เอกสาร Modern PetstoreAPI SSE.

การทดสอบ SSE ด้วย Apidog

Apidog รองรับการทดสอบ SSE:

สร้างคำขอ SSE
ตั้งค่า Accept: text/event-stream
เชื่อมต่อและดูเหตุการณ์แบบเรียลไทม์
ตรวจสอบรูปแบบเหตุการณ์
ทดสอบการเชื่อมต่อใหม่

แนวทางปฏิบัติที่ดีที่สุด

1. ตั้งค่า Headers ให้เหมาะสม

res.setHeader('Content-Type', 'text/event-stream');
res.setHeader('Cache-Control', 'no-cache');
res.setHeader('Connection', 'keep-alive');
res.setHeader('X-Accel-Buffering', 'no'); // Disable nginx buffering

2. ส่ง Heartbeats

รักษาการเชื่อมต่อให้คงอยู่:

const heartbeat = setInterval(() => {
  res.write(': heartbeat\n\n');
}, 15000);

res.on('close', () => clearInterval(heartbeat));

3. จัดการข้อผิดพลาดอย่างเหมาะสม

eventSource.onerror = (error) => {
  if (eventSource.readyState === EventSource.CLOSED) {
    // Connection closed, will auto-reconnect
  } else {
    // Other error
    console.error('SSE error:', error);
  }
};

4. ใช้ Event IDs สำหรับการดำเนินการต่อ

let lastEventId = 0;

app.get('/stream', (req, res) => {
  const startId = parseInt(req.headers['last-event-id'] || '0');

  for (let i = startId + 1; i <= 100; i++) {
    res.write(`id: ${i}\ndata: {"message":"Event ${i}"}\n\n`);
  }
});

5. ปิดการเชื่อมต่อ

// Client
eventSource.addEventListener('complete', () => {
  eventSource.close();
});

// Server
res.on('close', () => {
  // Cleanup resources
});

สรุป

SSE เหมาะอย่างยิ่งสำหรับการสตรีมการตอบสนอง API มันง่ายกว่า WebSocket สำหรับการสื่อสารทางเดียว ทำงานผ่าน HTTP และจัดการการเชื่อมต่อใหม่โดยอัตโนมัติ

Modern PetstoreAPI ใช้ SSE สำหรับการสตรีม AI, การอัปเดตคำสั่งซื้อ และฟีดสด ทดสอบปลายทาง SSE ด้วย Apidog

คำถามที่พบบ่อย

SSE สามารถทำงานผ่านไฟร์วอลล์ขององค์กรได้หรือไม่?

ได้ SSE ใช้ HTTP/HTTPS มาตรฐาน จึงสามารถทำงานผ่านไฟร์วอลล์และพร็อกซีส่วนใหญ่ได้

การเชื่อมต่อ SSE สามารถเปิดค้างไว้ได้นานแค่ไหน?

ได้ไม่จำกัด แต่ควรใช้ heartbeats ทุกๆ 15-30 วินาที เพื่อรักษาการเชื่อมต่อให้คงอยู่ผ่านพร็อกซี

ฉันสามารถส่งข้อมูลไบนารีผ่าน SSE ได้หรือไม่?

ไม่ได้ SSE เป็นแบบข้อความเท่านั้น ควรเข้ารหัสข้อมูลไบนารีเป็น Base64 หรือใช้ WebSocket แทน

SSE รองรับการสื่อสารแบบสองทางหรือไม่?

ไม่ได้ SSE เป็นแบบเซิร์ฟเวอร์ถึงไคลเอนต์เท่านั้น ไคลเอนต์ใช้คำขอ HTTP ปกติสำหรับการสื่อสารจากไคลเอนต์ถึงเซิร์ฟเวอร์

เบราว์เซอร์สามารถมีการเชื่อมต่อ SSE ได้กี่รายการ?

เบราว์เซอร์จำกัดการเชื่อมต่อ SSE ต่อโดเมน (โดยทั่วไปคือ 6) ใช้ multiplexing หรือ WebSocket สำหรับการเชื่อมต่อจำนวนมาก