פיתוח פלטפורמת סטרימינג בשידור חי מאפס

ארכיטקטורה של פלטפורמת סטרימינג בשידור חי

לעתים קרובות אנו נתקלים באתגר: חברה רוצה להשיק שירות סטרימינג משלה, דומה ל-Twitch, אך עם כללי מונטיזציה ותוכן משלה. בעיה אופיינית היא חביון מוגבר שמגיע עד 10 שניות בשעות שיא, ירידת איכות, ותשתית שלא יכולה לעמוד בעומס. במשך 5+ שנים פיתחנו יותר מ-15 פרויקטי סטרימינג, מפלטפורמות חינוכיות קטנות ועד מדיה גדולה. בואו נפרק כיצד לבנות מערכת חיה אמינה מאפס, תוך התחשבות בדרישות חביון, תאימות מכשירים וקנה מידה לאלפי צופים. בחירת הפרוטוקולים מבוססת על קריטריונים כמו חביון, תמיכת CDN ותאימות דפדפנים.

תכנית אספקה ופרוטוקולים

סטרימר → קליטה → טרנסקודינג → CDN → צופה

הזרם הנכנס מהסטרימר הוא בדרך כלל RTMP (OBS, StreamLabs, XSplit כולם תומכים ב-RTMP כברירת מחדל). בצד הצופה, אנו משתמשים ב-HLS או DASH לדפדפנים, WebRTC לחביון נמוך במיוחד (< 1 שנייה).

RTMP מספק חביון נמוך במהלך הקליטה אך אינו מתאים לאספקה עקב חסימת חומות אש. HLS הוא התקן דה פקטו, עובד דרך HTTP ונשמר בקלות במטמון ב-CDN.

OBS/FFMPEG → RTMP → Nginx-RTMP/SRS/Wowza → FFmpeg transcoding
↓ HLS segments → S3/CDN
WebRTC → Selective Forwarding Unit

למה SRS עדיף על שרתי קליטה אחרים

SRS (Simple Realtime Server) הוא קוד פתוח, כתוב ב-Go, ומטפל בעד 10K+ חיבורים בשרת יחיד. בפרויקטים שלנו, SRS הראה ביצועים גבוהים ב-30% מ-Wowza בתצורה זהה. הגדרה דרך קובץ תצורה יחיד:

# srs.conf
listen 1935;
max_connections 1000;
daemon off;
http_server {
    enabled on;
    listen 8080;
    dir ./objs/nginx/html;
}
vhost __defaultVhost__ {
    # Хук: уведомляем backend о начале/конце трансляции
    http_hooks {
        enabled on;
        on_publish http://api:8000/hooks/stream/start;
        on_unpublish http://api:8000/hooks/stream/stop;
        on_play http://api:8000/hooks/stream/view;
    }
    hls {
        enabled on;
        hls_path ./objs/nginx/html;
        hls_fragment 2; # 2 секунды — баланс задержки и стабильности
        hls_window 10; # 10 сегментов в окне
    }
    transcode {
        enabled on;
        ffmpeg /usr/local/bin/ffmpeg;
        engine hd {
            enabled on;
            vcodec libx264;
            vbitrate 2000;
            vfps 30;
            vwidth 1280;
            vheight 720;
            acodec aac;
            abitrate 128;
            output rtmp://localhost:1935/[app]/[stream]_720p;
        }
        engine sd {
            enabled on;
            vcodec libx264;
            vbitrate 800;
            vfps 30;
            vwidth 854;
            vheight 480;
            acodec aac;
            abitrate 96;
            output rtmp://localhost:1935/[app]/[stream]_480p;
        }
    }
}

אימות סטרימר

הסטרימר מפרסם את הזרם באמצעות מפתח סטרים. אסור לנו לקבל RTMP ממקורות לא ידועים:

# FastAPI: хук для SRS on_publish
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

class PublishHook(BaseModel):
    action: str
    app: str
    stream: str  # stream key от стримера
    param: str  # query string

@app.post("/hooks/stream/start")
async def on_stream_start(hook: PublishHook):
    # Валидируем stream key
    streamer = await db.fetchrow(
        "SELECT id, user_id, is_active FROM stream_keys WHERE key = $1",
        hook.stream
    )
    if not streamer or not streamer['is_active']:
        raise HTTPException(status_code=403, detail="Invalid stream key")

    # Запускаем трансляцию в БД
    await db.execute("""
        INSERT INTO live_streams (user_id, stream_key_id, started_at, status)
        VALUES ($1, $2, NOW(), 'live')
        ON CONFLICT (stream_key_id) DO UPDATE SET started_at = NOW(), status = 'live'
    """, streamer['user_id'], streamer['id'])

    # Уведомляем подписчиков через WebSocket
    await notify_followers(streamer['user_id'], 'stream_started')

    return {"code": 0}  # SRS ожидает code=0 для разрешения

איך עובד צ'אט בזמן אמת

צ'אט סטרים הוא חובה. WebSocket דרך Redis Pub/Sub עם הגבלת קצב וחלון החלקה של הודעות:

// Node.js: WebSocket-сервер для чата
import { WebSocketServer } from 'ws';
import { createClient } from 'redis';

const wss = new WebSocketServer({ port: 3001 });
const redis = createClient({ url: process.env.REDIS_URL });
const redisSub = redis.duplicate();

await redis.connect();
await redisSub.connect();

interface ChatMessage {
  type: 'message' | 'emote' | 'sub' | 'ban';
  streamId: string;
  userId: string;
  username: string;
  text: string;
  badges: string[];
  timestamp: number;
}

// Подписываемся на канал трансляции
wss.on('connection', (ws, req) => {
  const streamId = new URL(req.url!, 'ws://x').searchParams.get('stream');
  if (!streamId) return ws.close();

  const channel = `chat:${streamId}`;

  // Слушаем Redis Pub/Sub для этого стрима
  redisSub.subscribe(channel, (message) => {
    if (ws.readyState === ws.OPEN) {
      ws.send(message);
    }
  });

  ws.on('message', async (data) => {
    const msg: ChatMessage = JSON.parse(data.toString());

    // Антиспам: rate limit на пользователя
    const key = `chat_limit:${msg.userId}:${streamId}`;
    const count = await redis.incr(key);
    if (count === 1) await redis.expire(key, 5);
    if (count > 20) {
      // 20 сообщений за 5 секунд — слишком много
      ws.send(JSON.stringify({ type: 'slowmode', waitMs: 5000 }));
      return;
    }

    // Сохраняем в Redis Stream (скользящее окно 1000 сообщений)
    await redis.xAdd(`stream_chat:${streamId}`, '*', msg as any, {
      TRIM: { strategy: 'MAXLEN', threshold: 1000 }
    });

    // Публикуем всем подключённым
    await redis.publish(channel, JSON.stringify(msg));
  });

  ws.on('close', () => {
    redisSub.unsubscribe(channel);
  });
});

הקלטת סטרים ל-VOD

לאחר סיום הסטרים, אנו מאחדים את מקטעי ה-TS ומקודדים מחדש עם faststart לפסאודו-סטרימינג. תיעוד FFmpeg הרשמי ממליץ להשתמש בדגל OBS/FFMPEG → RTMP → Nginx-RTMP/SRS/Wowza → FFmpeg transcoding ↓ HLS segments → S3/CDN WebRTC → Selective Forwarding Unit כדי להעביר את אטום ה-moov לתחילת הקובץ, מה שמאיץ את תחילת ההפעלה.

# Celery task: конвертация в VOD
@app.task
def process_vod(stream_id: int):
    stream = LiveStream.objects.get(id=stream_id)
    segments = sorted(
        glob(f"/var/srs/hls/{stream.stream_key}/*.ts"),
        key=lambda f: int(Path(f).stem.split('_')[-1])
    )
    concat_list = "/tmp/vod_concat.txt"
    with open(concat_list, 'w') as f:
        for s in segments:
            f.write(f"file '{s}'\n")
    raw_mp4 = f"/tmp/vod_{stream_id}_raw.mp4"
    subprocess.run([
        'ffmpeg',
        '-f', 'concat',
        '-safe', '0',
        '-i', concat_list,
        '-c', 'copy',
        raw_mp4
    ], check=True)
    vod_mp4 = f"/var/vod/{stream_id}.mp4"
    subprocess.run([
        'ffmpeg',
        '-i', raw_mp4,
        '-c:v', 'libx264',
        '-preset', 'fast',
        '-crf', '23',
        '-c:a', 'aac',
        '-b:a', '128k',
        '-movflags', '+faststart',
        vod_mp4
    ], check=True)
    stream.vod_path = vod_mp4
    stream.status = 'ended'
    stream.save()

כיצד להגדיר טרנסקודינג מרובה איכויות: שלב אחר שלב

1. התקן SRS ו-FFmpeg על השרת.
2. בתצורת SRS, הפעל את הקטע # srs.conf listen 1935; max_connections 1000; daemon off; http_server { enabled on; listen 8080; dir ./objs/nginx/html; } vhost __defaultVhost__ { # Хук: уведомляем backend о начале/конце трансляции http_hooks { enabled on; on_publish http://api:8000/hooks/stream/start; on_unpublish http://api:8000/hooks/stream/stop; on_play http://api:8000/hooks/stream/view; } hls { enabled on; hls_path ./objs/nginx/html; hls_fragment 2; # 2 секунды — баланс задержки и стабильности hls_window 10; # 10 сегментов в окне } transcode { enabled on; ffmpeg /usr/local/bin/ffmpeg; engine hd { enabled on; vcodec libx264; vbitrate 2000; vfps 30; vwidth 1280; vheight 720; acodec aac; abitrate 128; output rtmp://localhost:1935/[app]/[stream]_720p; } engine sd { enabled on; vcodec libx264; vbitrate 800; vfps 30; vwidth 854; vheight 480; acodec aac; abitrate 96; output rtmp://localhost:1935/[app]/[stream]_480p; } } } וציין את הנתיבים ל-FFmpeg.
3. הגדר פרופילי טרנסקודינג (לדוגמה, HD: 720p, 30fps, 2 Mbps; SD: 480p, 30fps, 800 Kbps).
4. בכתובות הפלט, השתמש ב-# FastAPI: хук для SRS on_publish from fastapi import FastAPI, HTTPException from pydantic import BaseModel class PublishHook(BaseModel): action: str app: str stream: str # stream key от стримера param: str # query string @app.post("/hooks/stream/start") async def on_stream_start(hook: PublishHook): # Валидируем stream key streamer = await db.fetchrow( "SELECT id, user_id, is_active FROM stream_keys WHERE key = $1", hook.stream ) if not streamer or not streamer['is_active']: raise HTTPException(status_code=403, detail="Invalid stream key") # Запускаем трансляцию в БД await db.execute(""" INSERT INTO live_streams (user_id, stream_key_id, started_at, status) VALUES ($1, $2, NOW(), 'live') ON CONFLICT (stream_key_id) DO UPDATE SET started_at = NOW(), status = 'live' """, streamer['user_id'], streamer['id']) # Уведомляем подписчиков через WebSocket await notify_followers(streamer['user_id'], 'stream_started') return {"code": 0} # SRS ожидает code=0 для разрешения ו-// Node.js: WebSocket-сервер для чата import { WebSocketServer } from 'ws'; import { createClient } from 'redis'; const wss = new WebSocketServer({ port: 3001 }); const redis = createClient({ url: process.env.REDIS_URL }); const redisSub = redis.duplicate(); await redis.connect(); await redisSub.connect(); interface ChatMessage { type: 'message' | 'emote' | 'sub' | 'ban'; streamId: string; userId: string; username: string; text: string; badges: string[]; timestamp: number; } // Подписываемся на канал трансляции wss.on('connection', (ws, req) => { const streamId = new URL(req.url!, 'ws://x').searchParams.get('stream'); if (!streamId) return ws.close(); const channel = `chat:${streamId}`; // Слушаем Redis Pub/Sub для этого стрима redisSub.subscribe(channel, (message) => { if (ws.readyState === ws.OPEN) { ws.send(message); } }); ws.on('message', async (data) => { const msg: ChatMessage = JSON.parse(data.toString()); // Антиспам: rate limit на пользователя const key = `chat_limit:${msg.userId}:${streamId}`; const count = await redis.incr(key); if (count === 1) await redis.expire(key, 5); if (count > 20) { // 20 сообщений за 5 секунд — слишком много ws.send(JSON.stringify({ type: 'slowmode', waitMs: 5000 })); return; } // Сохраняем в Redis Stream (скользящее окно 1000 сообщений) await redis.xAdd(`stream_chat:${streamId}`, '*', msg as any, { TRIM: { strategy: 'MAXLEN', threshold: 1000 } }); // Публикуем всем подключённым await redis.publish(channel, JSON.stringify(msg)); }); ws.on('close', () => { redisSub.unsubscribe(channel); }); }); כך ש-SRS יוסיף את הסיומת אוטומטית.
5. ודא שמקטעי HLS נוצרים עבור כל פרופיל בתיקיות משנה נפרדות.
6. בדוק עם OBS, שלח את הזרם לקליטת RTMP. ודא שהנגן יכול לעבור בין רמות איכות.

השוואת פרוטוקולי אספקה

CDN פופולריים לאספקת HLS: השוואה

מה כלול בפיתוח פלטפורמה במפתח מלא

• עיצוב ארכיטקטוני: בחירת פרוטוקולים, CDN, תכנון קיבולת.
• שרת קליטה: הגדרת SRS/Nginx-RTMP, טרנסקודינג מרובה איכויות.
• צד שרת: API לניהול סטרים, אימות, מונטיזציה.
• צד לקוח: נגן מותאם אישית, לוח בקרה לסטרימר, צ'אט.
• תשתית: קונטיינריזציה עם Docker, קנה מידה אוטומטי, ניטור.
• תיעוד: תיעוד טכני מלא, הוראות למנהלים.
• הדרכה: מפגש לצוות שלך, שבועיים של תמיכה לאחר השקה.
• אחריות: תיקוני באגים בחינם לחודש אחד לאחר ההשקה.

קנה מידה: קליטה מרובת שרתים

שרת קליטה יחיד הוא נקודת כשל יחידה. לייצור, אתה צריך קלאסטר עם איזון עומסים:

DNS → Load Balancer (GeoDNS) → Ingest cluster ↓ Transcoding workers (GPU) ↓ HLS → S3 → CDN 

סטרימרים מופנים לשרת הקליטה הקרוב ביותר דרך GeoDNS. כל שרת קליטה כותב לאחסון אובייקטים משותף או משכפל מקטעים באופן סינכרוני. צור קשר כדי לדון בארכיטקטורת הפרויקט שלך — נמצא את התצורה האופטימלית לצרכים שלך.

לוחות זמנים

MVP עם קליטת RTMP, אספקת HLS, צ'אט WebSocket והקלטת VOD — 10–12 שבועות. הוספת טרנסקודינג מרובה איכויות, מנויי מתנה, ניהול צ'אט, נגן נייד — עוד 8–10 שבועות. קנה מידה ל-10k+ צופים במקביל, קליטה מאוזנת עומסים, CDN עם origin shield — שלב נפרד.

מוכן להשיק שירות סטרימינג משלך? קבל ייעוץ — נעריך את הפרויקט שלך תוך יומיים. צור קשר.