WebSocket على Rust: إنشاء تطبيقات الوقت الفعلي المتقدمة

مقدمة: لماذا Rust لـ WebSocket؟

تتطلب تطبيقات الوقت الفعلي الحديثة - الدردشات، محطات التداول، الألعاب عبر الإنترنت، أنظمة المراقبة - أداءً عاليًا، تأخيرًا ضئيلًا، وموثوقية. Rust مع نظام الملكية الخاص بها، وتكلفة الصفر للتجريدات، وبيئة التشغيل غير المتزامنة Tokio تصبح الخيار المثالي لبناء خوادم WebSocket قابلة للتوسع.

في هذه المقالة، سنتجاوز حدود خادم الصدى البسيط وننظر في التقنيات المتقدمة: إدارة آلاف الاتصالات المتزامنة، تجمعات الخيوط، أطر العمل لبناء منطق معقد، بالإضافة إلى أنماط التوسع الأفقي. المادة موجهة للمطورين الذين لديهم بالفعل معرفة بأساسيات Rust والبرمجة غير المتزامنة.

هندسة خادم WebSocket عالي التحميل

1. الأساس غير المتزامن: Tokio و warp

للمستوى الإنتاجي، نستخدم مجموعة Tokio (بيئة تشغيل غير متزامنة) و warp (إطار ويب قائم على المرشحات). على عكس الحلول البسيطة، يوفر warp تركيبة مرنة للمسارات ودعمًا مدمجًا لـ WebSocket.

use warp::Filter;use futures_util::{SinkExt, StreamExt};use tokio::sync::broadcast;

#[tokio::main]async fn main() {    let (tx, _rx) = broadcast::channel::<String>(100);        let ws_route = warp::path("ws")        .and(warp::ws())        .and(with_sender(tx.clone()))        .map(|ws: warp::ws::Ws, tx| {            ws.on_upgrade(move |socket| handle_connection(socket, tx))        });

    let health_route = warp::path("health")        .map(|| warp::reply::json(&serde_json::json!({"status": "ok"})));

    let routes = ws_route.or(health_route);    warp::serve(routes).run(([0, 0, 0, 0], 8080)).await;}

fn with_sender(tx: broadcast::Sender<String>) -> impl Filter<Extract = (broadcast::Sender<String>,), Error = std::convert::Infallible> + Clone {    warp::any().map(move || tx.clone())}

النقاط الرئيسية:

• نستخدم broadcast::channel لإرسال الرسائل إلى جميع العملاء المتصلين (نمط "pub/sub").
• القناة بسعة 100 رسالة تمنع الحظر عند حدوث زيادة مفاجئة في حركة المرور.
• المرشح with_sender يقوم بحقن المرسل في المعالج من خلال الإغلاق.

2. إدارة حالة الاتصالات

على المستوى المتقدم، لا يكفي مجرد إرسال الرسائل - بل يجب إدارة دورة حياة كل اتصال، وتتبع الغرف، والمصادقة، والمهلات الزمنية. نستخدم HashMap مع Arc<Mutex<>> للوصول الآمن من مهام متعددة.

use std::collections::HashMap;use std::sync::Arc;use tokio::sync::Mutex;use warp::ws::Message;

type Clients = Arc<Mutex<HashMap<String, Vec<tokio::sync::mpsc::UnboundedSender<Message>>>>>;

async fn handle_connection(ws: warp::ws::WebSocket, clients: Clients, room_id: String) {    let (mut ws_tx, mut ws_rx) = ws.split();    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Message>();        // تسجيل العميل في الغرفة    {        let mut map = clients.lock().await;        map.entry(room_id.clone()).or_insert_with(Vec::new).push(tx);    }

    // مهمة إرسال الرسائل إلى العميل    let send_task = tokio::spawn(async move {        while let Some(msg) = rx.recv().await {            if ws_tx.send(msg).await.is_err() {                break;            }        }    });

    // مهمة استقبال الرسائل من العميل    let receive_task = tokio::spawn(async move {        while let Some(Ok(msg)) = ws_rx.next().await {            // هنا يمكن معالجة الرسالة الواردة            // على سبيل المثال، تحليل JSON، التحقق من الصلاحيات            println!("Received: {:?}", msg);        }    });

    // انتظار انتهاء أي من