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

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

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

هندسة خادم WebSocket باستخدام Tokio و Tungstenite

لبناء خادم موثوق، نستخدم مجموعة tokio (بيئة تشغيل غير متزامنة) و tokio-tungstenite (مكتبة WebSocket). تتضمن الهندسة المتقدمة تجمع اتصالات، إيقاف تشغيل سلس، ومعالجة الأخطاء.

use tokio::net::TcpListener;\nuse tokio_tungstenite::accept_async;\nuse futures_util::{SinkExt, StreamExt};\nuse std::sync::Arc;\nuse tokio::sync::Mutex;\n\ntype PeerMap = Arc<Mutex<HashMap<String, PeerConnection>>>;\n\n#[derive(Clone)]\nstruct PeerConnection {\n    sender: mpsc::UnboundedSender<Message>,\n    user_id: String,\n}\n\nasync fn handle_connection(\n    peer_map: PeerMap,\n    raw_stream: TcpStream,\n    addr: SocketAddr,\n) {\n    let ws_stream = accept_async(raw_stream)\n        .await\n        .expect("خطأ أثناء المصافحة");\n    \n    let (mut write, mut read) = ws_stream.split();\n    let (tx, mut rx) = mpsc::unbounded_channel::<Message>();\n    \n    // التسجيل في التجمع\n    let user_id = Uuid::new_v4().to_string();\n    peer_map.lock().await.insert(\n        user_id.clone(),\n        PeerConnection { sender: tx, user_id: user_id.clone() }\n    );\n    \n    // تشغيل المهام\n    let send_task = tokio::spawn(async move {\n        while let Some(msg) = rx.recv().await {\n            if write.send(msg).await.is_err() {\n                break;\n            }\n        }\n    });\n    \n    let recv_task = tokio::spawn(async move {\n        while let Some(Ok(msg)) = read.next().await {\n            broadcast_message(&peer_map, &user_id, &msg).await;\n        }\n    });\n    \n    // إيقاف تشغيل سلس\n    tokio::select! {\n        _ = send_task => {},\n        _ = recv_task => {},\n    }\n    \n    peer_map.lock().await.remove(&user_id);\n    println!("تم فصل العميل {}", addr);\n}\n

يوضح هذا الكود الهندسة الأساسية: تجمع اتصالات، اتصال ثنائي الاتجاه، وإزالة تلقائية عند الانقطاع.

التوسع: آلاف الاتصالات والتجميع العنقودي

لمعالجة آلاف الاتصالات، نستخدم عدة أنماط رئيسية:

\n
• التوسع الأفقي — تشغيل عدة نسخ من الخادم مع موازن تحميل
\n
• Redis Pub/Sub — تبادل الرسائل بين النسخ
\n
• التقسيم (Sharding) — التوزيع حسب user_id
\n

مثال على التكامل مع Redis للتبادل بين الخوادم:

use redis::AsyncCommands;\nuse serde::{Serialize, Deserialize};\n\n#[derive(Serialize, Deserialize)]\nstruct ClusterMessage {\n    from: String,\n    to: String,\n    payload: String,\n    timestamp: u64,\n}\n\nasync fn handle_cluster_message(\n    redis_conn: &mut redis::aio::Connection,\n    peer_map: PeerMap,\n    channel: String,\n) {\n    let mut pubsub = redis_conn.as_pubsub();\n    pubsub.subscribe(channel).await.unwrap();\n    \n    let mut stream = pubsub.on_message();\n    \n    while let Some(msg) = stream.next().await {\n        let payload: String = msg.get_payload().unwrap();\n        let cluster_msg: ClusterMessage = serde_json::from_str(&payload).unwrap();\n        \n        // إرسال إلى العميل المحلي\n        if let Some(peer) = peer_map.lock().await.get(&cluster_msg.to) {\n            let _ = peer.sender.send(Message::Text(cluster_msg.payload.into()));\n        }\n    }\n}\n

لتوزيع الحمل، نستخدم التجزئة المتسقة (consistent hashing). وهذا يضمن توزيعًا متساويًا لـ