Rust पर WebSocket: उन्नत real-time अनुप्रयोग

Rust पर WebSocket: उन्नत real-time अनुप्रयोग

Rust अपनी सुरक्षा और शून्य-लागत अमूर्तता के कारण उच्च-प्रदर्शन वाले real-time अनुप्रयोग बनाने के लिए एक आदर्श भाषा है। इस लेख में, हम WebSocket के साथ काम करने की उन्नत तकनीकों पर चर्चा करेंगे: स्केलिंग से लेकर प्रमाणीकरण और त्रुटि प्रबंधन तक। आप सीखेंगे कि कैसे एक production-ready WebSocket सर्वर बनाया जाए जो हजारों समवर्ती कनेक्शनों को संभाल सके।

Tokio और Tungstenite पर WebSocket सर्वर की आर्किटेक्चर

एक विश्वसनीय सर्वर बनाने के लिए हम tokio (एसिंक्रोनस रनटाइम) और tokio-tungstenite (WebSocket लाइब्रेरी) के संयोजन का उपयोग करते हैं। उन्नत आर्किटेक्चर में कनेक्शन पूल, graceful shutdown और त्रुटि प्रबंधन शामिल है।

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    // Graceful shutdown\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 का उपयोग करते हैं। यह समान वितरण सुनिश्चित करता है।