الأنظمة الفورية وCRDTs ومعالجة التدفق

ميزات الوقت الفعلي موجودة في كل مكان: المستندات التعاونية ولوحات المعلومات المباشرة والمحادثات والألعاب متعددة اللاعبين ومؤشرات الحضور. يغطي هذا الدرس البروتوكولات الأساسية وهياكل البيانات وأنماط المعالجة التي تشغّل هذه الأنظمة على نطاق واسع.

بروتوكولات النقل الفوري

توجد ثلاثة أساليب رئيسية لاتصال الخادم بالعميل. الاختيار الصحيح يعتمد على متطلبات زمن الاستجابة واتجاه الرسائل وقيود البنية التحتية.

Long Polling              SSE                    WebSocket
┌────────┐            ┌────────┐            ┌────────┐
│ العميل │            │ العميل │            │ العميل │
└───┬────┘            └───┬────┘            └───┬────┘
    │ GET /updates        │ GET /stream         │ Upgrade
    │────────────>        │────────────>        │────────────>
    │  (معلّق)            │  (معلّق)            │ (101 Switch)
    │                     │                     │
    │<──── JSON ──────    │<── data: {...} ──   │<──── frame ────
    │                     │<── data: {...} ──   │──── frame ───>
    │ GET /updates        │<── data: {...} ──   │<──── frame ────
    │────────────>        │                     │
    │  (تكرار)            │  (اتجاه واحد)       │ (ثنائي الاتجاه)

الميزة	Long Polling	SSE (أحداث مرسلة من الخادم)	WebSocket
الاتجاه	خادم إلى عميل (محاكاة)	خادم إلى عميل فقط	ثنائي الاتجاه
البروتوكول	HTTP/1.1 طلبات متكررة	HTTP/1.1 مع text/event-stream	TCP مُرقّى عبر ws:// أو wss://
عبء الاتصال	مصافحة TCP جديدة لكل استطلاع	اتصال مستمر واحد	اتصال مستمر واحد
البيانات الثنائية	لا (JSON/نص فقط)	لا (UTF-8 نص فقط)	نعم (إطارات ثنائية مدعومة)
إعادة الاتصال التلقائي	تنفيذ يدوي	مدمج (واجهة EventSource)	تنفيذ يدوي
توافق الوكلاء/CDN	نعم (HTTP قياسي)	نعم (HTTP قياسي)	غالبًا يتطلب تكوين
مخاوف التوسع	حجم طلبات عالٍ	اتصال واحد لكل عميل	اتصال واحد لكل عميل
الأفضل لـ	الأنظمة القديمة، الإشعارات البسيطة	البث المباشر، لوحات المعلومات، مؤشرات الأسهم	المحادثات، التعاون، الألعاب

قاعدة عامة للمقابلة: استخدم SSE عندما يدفع الخادم البيانات باتجاه واحد (لوحات المعلومات، الإشعارات). استخدم WebSocket عندما يرسل العميل أيضًا بيانات متكررة (التحرير التعاوني، المحادثات). تجنب long polling في التصاميم الجديدة إلا عند وجود قيود بنية تحتية تمنع الاتصالات المستمرة.

إدارة اتصالات WebSocket على نطاق واسع

خادم واحد يمكنه الاحتفاظ بعشرات الآلاف من اتصالات WebSocket، لكن على نطاق واسع تواجه تحديات التوجيه وتجاوز الفشل وإدارة الحالة.

استراتيجيات توجيه الاتصالات

                    موازن الحمل
                    ┌───────────┐
  العميل A ────────>│           │──────> الخادم 1 (يحمل A, B)
  العميل B ────────>│  L7 LB    │──────> الخادم 2 (يحمل C, D)
  العميل C ────────>│           │──────> الخادم 3 (يحمل E, F)
  العميل D ────────>│           │
  العميل E ────────>│           │
  العميل F ────────>└───────────┘

الجلسات اللاصقة (Sticky sessions) تربط العميل بنفس الخادم باستخدام ملف تعريف ارتباط أو تجزئة معرّف العميل. بسيطة التنفيذ، لكنها تسبب مشاكل أثناء النشر المتدرج (الاتصالات تنقطع) وتوزيع حمل غير متساوٍ.

ترحيل الاتصال (Connection migration) يخزن حالة الاتصال خارجيًا (في Redis) حتى يتمكن أي خادم من استئناف الجلسة. عندما يتوقف خادم، يعيد العملاء الاتصال بأي خادم متاح يحمّل حالتهم من Redis. أكثر تعقيدًا لكن يزيل عيوب الجلسات اللاصقة.

نبضات القلب وإعادة الاتصال

import asyncio
import time
from dataclasses import dataclass, field

@dataclass
class ConnectionState:
    client_id: str
    last_heartbeat: float = field(default_factory=time.time)
    missed_heartbeats: int = 0

class HeartbeatManager:
    def __init__(self, interval_sec: float = 30.0, max_missed: int = 3):
        self.interval = interval_sec
        self.max_missed = max_missed
        self.connections: dict[str, ConnectionState] = {}

    def register(self, client_id: str) -> None:
        self.connections[client_id] = ConnectionState(client_id=client_id)

    def heartbeat_received(self, client_id: str) -> None:
        if client_id in self.connections:
            conn = self.connections[client_id]
            conn.last_heartbeat = time.time()
            conn.missed_heartbeats = 0

    def check_connections(self) -> list[str]:
        """إرجاع قائمة معرّفات العملاء التي يجب قطع اتصالها."""
        now = time.time()
        dead: list[str] = []
        for client_id, conn in self.connections.items():
            if now - conn.last_heartbeat > self.interval:
                conn.missed_heartbeats += 1
                if conn.missed_heartbeats >= self.max_missed:
                    dead.append(client_id)
        return dead

يجب على العملاء تنفيذ تراجع أسي مع اهتزاز (jitter) عند إعادة الاتصال لتجنب القطيع المتدافع عند إعادة تشغيل الخادم:

import random

def reconnect_delay(attempt: int, base_ms: int = 1000, max_ms: int = 30000) -> float:
    """تراجع أسي: 1 ثانية، 2، 4، 8... حد أقصى 30 ثانية، مع اهتزاز."""
    delay = min(base_ms * (2 ** attempt), max_ms)
    jitter = random.uniform(0, delay * 0.3)  # اهتزاز 0-30%
    return (delay + jitter) / 1000.0

بنية Pub/Sub

نمط Pub/Sub يفصل المنتجين عن المستهلكين، مما يتيح التوزيع لعدة مشتركين بدون اتصالات نقطة لنقطة.

Kafka مقابل Redis Streams

الجانب	Kafka	Redis Streams
نموذج التخزين	سجل إلحاقي مقسّم على القرص	تدفق في الذاكرة مع استمرارية اختيارية
الاحتفاظ	قابل للتكوين (أيام/أسابيع/للأبد)	محدود بالذاكرة أو بطول أقصى
مجموعات المستهلكين	نعم (تعيين مبني على التقسيمات)	نعم (XREADGROUP مع إقرار)
ضمان الترتيب	FIFO لكل تقسيم	FIFO لكل تدفق
الإنتاجية	ملايين الرسائل/ثانية (إدخال/إخراج مجمّع)	مئات الآلاف/ثانية (خيط واحد)
زمن الاستجابة	ميللي ثوانٍ قليلة (التجميع يضيف تأخيرًا صغيرًا)	أقل من ميللي ثانية
الأفضل لـ	مصادر الأحداث، خطوط تحليل البيانات، السجلات المتينة	الإشعارات الفورية، pub/sub خفيف، أحداث طبقة التخزين المؤقت

أنماط التوزيع (Fan-Out)

التوزيع عند الكتابة (نموذج الدفع): عند نشر رسالة، يتم توصيلها فورًا لجميع المشتركين. يستخدمه Twitter للمستخدمين ذوي المتابعين القليلين. قراءات سريعة، كتابات مكلفة.

التوزيع عند القراءة (نموذج السحب): المشتركون يستعلمون عن الرسائل الجديدة عند الطلب. يستخدمه Twitter لحسابات المشاهير (ملايين المتابعين). كتابات رخيصة، قراءات أبطأ.

التوزيع عند الكتابة                التوزيع عند القراءة

المنتج                              المنتج
   │                                   │
   ▼                                   ▼
 نشر ──> صندوق المشترك A            سجل الموضوع
       ──> صندوق المشترك B              │
       ──> صندوق المشترك C           ┌──┼──┐
                                     ▼  ▼  ▼
  (N كتابة لكل رسالة)             A   B   C  (قراءة عند الطلب)
                                   (N قراءة لكل مستهلك)

رؤية للمقابلة: معظم الأنظمة تستخدم نهجًا هجينًا. خدمة التوزيع في Twitter تدفع للجداول الزمنية للمستخدمين ذوي أقل من 500 متابع لكنها تخزن تغريدات المشاهير في الجدول الزمني الرئيسي للدمج بالسحب.

CRDTs للتحرير التعاوني

عندما يحرر عدة مستخدمين نفس المستند في وقت واحد، تحتاج استراتيجية لدمج التغييرات المتزامنة بدون تعارضات.

OT مقابل CRDTs

التحويل التشغيلي (OT) يحوّل العمليات ضد بعضها للحفاظ على الاتساق. Google Docs يستخدم OT مع خادم مركزي يرتب العمليات.

أنواع البيانات المنسوخة الخالية من التعارض (CRDTs) هي هياكل بيانات مصممة بحيث تتقارب التحديثات المتزامنة دائمًا بدون تنسيق. Figma انتقلت من OT إلى CRDTs في عام 2019، موثقة علنيًا الفوائد لأداة التصميم متعددة اللاعبين.

الجانب	OT (Google Docs)	CRDTs (Figma, منذ 2019)
خادم مركزي مطلوب	نعم (يرتب العمليات)	لا (ند لند ممكن)
التعقيد	دوال تحويل O(n^2)	دمج O(1) لكل عملية
دعم عدم الاتصال	محدود (يحتاج خادمًا)	كامل (دمج عند إعادة الاتصال)
إثبات الصحة	صعب (حالات حدية كثيرة)	ضمان رياضي
عبء الذاكرة	منخفض (عمليات فقط)	أعلى (بيانات وصفية لكل عنصر)

CRDTs العدادات

G-Counter (عداد النمو فقط): كل عقدة تحتفظ بعدادها الخاص. القيمة هي مجموع جميع العقد. الدمج يأخذ الحد الأقصى لعداد كل عقدة.

from typing import Dict

class GCounter:
    """عداد CRDT للنمو فقط. يدعم الزيادة والدمج."""
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.counts: Dict[str, int] = {node_id: 0}

    def increment(self, amount: int = 1) -> None:
        self.counts[self.node_id] = self.counts.get(self.node_id, 0) + amount

    def value(self) -> int:
        return sum(self.counts.values())

    def merge(self, other: "GCounter") -> None:
        for node_id, count in other.counts.items():
            self.counts[node_id] = max(self.counts.get(node_id, 0), count)

PN-Counter (عداد موجب-سالب): عدادي G-Counter — واحد للزيادات وواحد للنقصانات. القيمة = P.value() - N.value().

LWW-Register (سجل آخر كاتب يفوز)

يخزن قيمة واحدة مع طابع زمني. عند التعارض، الطابع الزمني الأعلى يفوز.

from dataclasses import dataclass
from typing import Any

@dataclass
class LWWRegister:
    """سجل آخر كاتب يفوز. الطابع الزمني الأعلى يفوز دائمًا."""
    value: Any = None
    timestamp: float = 0.0
    node_id: str = ""

    def set(self, value: Any, timestamp: float, node_id: str) -> None:
        if timestamp > self.timestamp or (
            timestamp == self.timestamp and node_id > self.node_id
        ):
            self.value = value
            self.timestamp = timestamp
            self.node_id = node_id

    def merge(self, other: "LWWRegister") -> None:
        if other.timestamp > self.timestamp or (
            other.timestamp == self.timestamp and other.node_id > self.node_id
        ):
            self.value = other.value
            self.timestamp = other.timestamp
            self.node_id = other.node_id

OR-Set (مجموعة الإضافة-الحذف المرصود)

تدعم الإضافة والحذف معًا. كل عنصر يُوسم بمعرّف فريد عند الإضافة. الحذف يزيل فقط الوسوم المحددة التي تمت ملاحظتها، لذا الإضافة المتزامنة لنفس العنصر تُحفظ.

CRDTs النصية: RGA و LSEQ

للتحرير التعاوني للنصوص، تحتاج CRDTs تنمذج تسلسلات الأحرف:

• RGA (المصفوفة القابلة للنمو المنسوخة): كل حرف له معرّف فريد (node_id, sequence). الإدراج يضع حرفًا بعد موضع مرجعي. الحذف يُعلّم الأحرف كشواهد. الإدراجات المتزامنة في نفس الموضع تُرتب بشكل حتمي بمعرّف العقدة.

• LSEQ: يُعيّن لكل حرف موضعًا من فضاء معرّفات كثيف. المواضع تُخصص بين الجيران الحاليين، متجنبة إعادة التوازن. أكثر كفاءة في المساحة من RGA للمستندات الكبيرة.

  مثال إدراج RGA (تحريرات متزامنة):

  البداية:     H - E - L - L - O

  المستخدم A يُدرج 'X' بعد 'E':   H - E - X - L - L - O
  المستخدم B يُدرج 'Y' بعد 'E':   H - E - Y - L - L - O

  بعد الدمج (A.id > B.id):        H - E - X - Y - L - L - O
  كلا الإدراجين محفوظان، ترتيب حتمي بمعرّف العقدة.

أنماط معالجة التدفق

التحليلات الفورية ومعالجة الأحداث تتطلب استراتيجيات تقسيم نوافذ لتجميع تدفقات البيانات غير المحدودة.

أنواع النوافذ

نافذة متقلبة (ثابتة، غير متداخلة):
|----5دقائق----|----5دقائق----|----5دقائق----|
|  أحداث       |  أحداث       |  أحداث       |

نافذة منزلقة (ثابتة، متداخلة):
|----5دقائق----|
   |----5دقائق----|
      |----5دقائق----|
(تتقدم كل دقيقة واحدة)

نافذة الجلسة (مبنية على الفجوة):
|--أحداث--|     |--أحداث------أحداث--|    |--أحداث--|
          ^فجوة^                     ^فجوة^
(النافذة تُغلق بعد فجوة خمول، مثلاً 30 دقيقة)

نوع النافذة	حالة الاستخدام	مثال
متقلبة	تجميع دوري	مشاهدات الصفحة لكل شريحة 5 دقائق
منزلقة	مقاييس متحركة	متوسط وقت الاستجابة خلال آخر 10 دقائق، يُحدّث كل دقيقة
الجلسة	سلوك المستخدم	إجمالي الإجراءات لكل جلسة مستخدم (تنتهي بعد 30 دقيقة خمول)

دلالات التسليم مرة واحدة بالضبط

في معالجة التدفق الموزعة، توجد ثلاثة ضمانات تسليم:

• مرة واحدة على الأكثر: أطلق وانسَ. الرسائل قد تُفقد. الأسرع.
• مرة واحدة على الأقل: أعد المحاولة عند الفشل. الرسائل قد تتكرر. يتطلب مستهلكين عديمي القوة.
• مرة واحدة بالضبط: كل رسالة تُعالج مرة واحدة فقط. يُحقق عبر كتابات عديمة القوة + إزاحات معاملاتية (معاملات Kafka) أو إزالة التكرار بمعرّفات أحداث فريدة.

العلامات المائية (Watermarks)

العلامة المائية هي تأكيد طابع زمني: "جميع الأحداث ذات الطابع الزمني <= W قد وصلت." العلامات المائية تتيح للنظام معرفة متى يمكن إغلاق النافذة وإصدار النتائج، حتى مع الأحداث غير المرتبة.

وقت الحدث:    10:00  10:01  10:03  10:02  10:04  10:05
               │      │      │      │      │      │
العلامة المائية: ──────────────────────────────────>
               W=10:00       W=10:02       W=10:04

حدث متأخر (10:02 يصل بعد W=10:02) يمكن:
- إسقاطه (الأبسط)
- وضعه في مخرج جانبي لإعادة المعالجة
- تشغيل تحديث النافذة (تأخر مسموح)

كشف الحضور على نطاق واسع

عرض حالة "متصل" لملايين المستخدمين يتطلب تصميمًا دقيقًا لتجنب إثقال النظام.

البنية

العميل ──نبضة كل 30 ثانية──> خدمة الحضور ──> Redis
                                     │
                                     ▼
                        تعيين TTL على المفتاح "user:{id}"
                        إلى 60 ثانية (ضعف فاصل النبضة)
                                     │
                                     ▼
                        المفتاح ينتهي تلقائيًا
                        إذا لم تُستلم نبضة
                        (المستخدم غير متصل)

استعلام الحالة:
  GET user:{id}  →  موجود? متصل : غير متصل

لعرض الحضور للأصدقاء/القنوات، تجنب الاستعلام لكل صديق. بدلاً من ذلك:

1. Pub/sub لتغييرات الحالة: عندما تتغير حالة المستخدم (متصل/غير متصل)، انشر في قناة. المشتركون (الأصدقاء المتصلون حاليًا) يستلمون التحديث.
2. التحميل الكسول: جلب الحضور فقط للمستخدمين المرئيين على الشاشة.
3. الاستعلامات المجمّعة: MGET user:101 user:102 user:103 في استدعاء Redis واحد.

تطبيق المقابلة: تصميم محرر مستندات فوري

عند طلب تصميم Google Docs أو محرر تعاوني في مقابلة، هيكل إجابتك:

1. النقل: WebSocket للتحرير الفوري ثنائي الاتجاه. رجوع إلى SSE + HTTP POST للشبكات المقيدة.
2. حل التعارضات: CRDTs (RGA للنصوص) للدمج التلقائي بدون خادم ترتيب مركزي. اذكر OT كبديل (نهج Google Docs) والمقايضات.
3. حالة المستند: كل حرف له معرّف CRDT فريد. العملاء يطبقون العمليات المحلية فورًا (متفائل) ويبثون للخادم. الخادم يدمج ويعيد البث.
4. الحضور: مواضع المؤشرات وحالة المستخدم تُتتبع عبر نبضات القلب مع مفاتيح TTL في Redis.
5. الاستمرارية: العمليات تُلحق بسجل (مصادر الأحداث). لقطات دورية للتحميل السريع للمستندات.
6. التوسع: تقسيم المستندات عبر الخوادم. كل مستند يعيش على خادم واحد (توجيه لاصق بمعرّف المستند). العمليات عبر المستندات نادرة.

التالي: اختبار ومختبر الوحدة 03 — ابنِ واجهة خلفية لمستند تعاوني مع CRDTs وإدارة WebSocket وتتبع الحضور. :::