使用Rust和WebSocket構(gòu)建點對點網(wǎng)絡(luò)
作者：李明
這篇文章使用Rust構(gòu)建基于WebSocket的P2P網(wǎng)絡(luò)示例，它提供了一種強大而有效的方式來實現(xiàn)節(jié)點之間的實時通信。通過理解代碼的每個部分，可以擴展和定制這個示例以滿足特定需求，無論是分散的應(yīng)用程序，實時數(shù)據(jù)共享還是分布式計算任務(wù)。?
在WebSocket基礎(chǔ)設(shè)施上創(chuàng)建點對點(P2P)網(wǎng)絡(luò)似乎是一項艱巨的任務(wù)。在這篇文章中，我們將介紹基于Rust的P2P網(wǎng)絡(luò)的關(guān)鍵組件，探索每個部分如何構(gòu)建無縫的WebSocket基礎(chǔ)設(shè)施。使用它們構(gòu)建一個健壯且高效的P2P網(wǎng)絡(luò)，允許節(jié)點之間的實時通信。
P2P網(wǎng)絡(luò)簡介
點對點網(wǎng)絡(luò)支持節(jié)點之間的分散通信，允許數(shù)據(jù)交換而不依賴于中心服務(wù)器。網(wǎng)絡(luò)中的每個參與者，或“對等端”，可以同時充當(dāng)客戶端和服務(wù)器端，促進直接連接和通信。在我們的示例中，我們使用WebSocket，它在單個TCP連接上提供全雙工通信通道，以促進這種實時交互。
項目概述
我們的項目旨在演示如何在Rust中使用WebSocket建立P2P網(wǎng)絡(luò)，利用Tokio異步運行時的強大功能。我們將探討以下關(guān)鍵組件：
- 命令行參數(shù)解析：使用clap解析對等url和綁定地址。
- WebSocket Actor：管理到對等端的WebSocket連接。
- 網(wǎng)絡(luò)狀態(tài)管理：維護網(wǎng)絡(luò)的狀態(tài)，包括連接的對等點。
- 連接處理：管理對等端WebSocket連接的生命周期。
- 廣播消息：定期向所有連接的對等端發(fā)送消息。
- 優(yōu)雅關(guān)閉：優(yōu)雅地處理中斷以關(guān)閉網(wǎng)絡(luò)。
項目開發(fā)
使用以下命令創(chuàng)建一個Rust新項目：
cargo new p2p-ws-example在Cargo.toml文件中加入以下依賴項：
[dependencies]
clap = { version = "4.5.13", features = ["derive"] }
env_logger = "0.11.5"
futures = "0.3.30"
log = "0.4.22"
serde = "1.0.204"
tokio = { version = "1.39.2", features = ["full"] }
tokio-tungstenite = "0.23.1"
tokio-util = { version = "0.7.11", features = ["full"] }命令行參數(shù)解析
我們將從使用clap定義命令行參數(shù)開始。這允許我們在啟動P2P節(jié)點時指定對等url和綁定地址。
use std::net::ToSocketAddrs;
use clap::Parser;
#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
/// 要連接的客戶端地址列表
#[arg(short, long, value_delimiter = ',', value_parser = parse_peer)]
peers: Vec<String>,
/// 綁定服務(wù)器的地址
#[arg(short, long, value_parser = parse_bind)]
bind: String,
}
/// 解析并驗證客戶端url
fn parse_peer(s: &str) -> Result<String, String> {
// 驗證以ws://或wss://開頭的URL
if s.starts_with("ws://") {
let ip_port = &s[5..];
if let Ok(_socket_addr) = ip_port.to_socket_addrs() {
return Ok(s.to_string());
}
}
Err(format!("Invalid client URL: {}", s))
}
/// 解析并驗證綁定地址
fn parse_bind(s: &str) -> Result<String, String> {
if let Ok(_socket_addr) = s.to_socket_addrs() {
return Ok(s.to_string());
}
Err(format!("Invalid bind address: {}", s))
}- peers：以逗號分隔的WebSocket url列表，以對等端的形式連接。
- bind：連接綁定服務(wù)器的地址。
WebSocket Actor
WebSocketActor結(jié)構(gòu)體管理WebSocket連接。它建立到給定URL的連接，并處理消息的發(fā)送和接收。
use tokio::net::TcpStream;
use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
// 定義 WebSocket actor
struct WebSocketActor {
ws_stream: WebSocketStream<MaybeTlsStream<TcpStream>>,
}
impl WebSocketActor {
async fn connect(url: &str) -> Option<Self> {
match connect_async(url).await {
Ok((conn, _)) => {
log::info!("Connected successfully to {}", url);
Some(WebSocketActor { ws_stream: conn })
}
Err(e) => {
log::error!("Connection to {} failed: {:?}", url, e);
None
}
}
}
}網(wǎng)絡(luò)狀態(tài)管理
P2PWebsocketNetwork結(jié)構(gòu)體維護網(wǎng)絡(luò)的狀態(tài)，包括連接的對等點和用于消息廣播的主發(fā)送方。
use std::{
collections::HashMap,
net::{SocketAddr, ToSocketAddrs},
sync::{Arc, Mutex},
};
use clap::Parser;
use tokio::{net::TcpStream, sync::mpsc::UnboundedSender};
use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
struct P2PWebsocketNetwork {
addresses: Arc<Mutex<HashMap<SocketAddr, UnboundedSender<P2PInnerMessage>>>>,
master: Arc<Mutex<UnboundedSender<P2PInnerMessage>>>,
}
#[derive(Debug)]
struct P2PInnerMessage {
message: Message,
tx_handler: UnboundedSender<P2PInnerMessage>,
}連接管理
我們將使用handle_connection和handle_server_connection函數(shù)處理傳入和傳出的連接。這些函數(shù)管理WebSocket連接的生命周期，發(fā)送和接收消息。
async fn handle_connection(
state: Arc<P2PWebsocketNetwork>,
conn: WebSocketActor,
token: CancellationToken,
) {
// 處理連接和消息交換的邏輯
}- state：包含網(wǎng)絡(luò)信息的共享狀態(tài)(Arc<P2PWebsocketNetwork>)，包括已連接的對等端和消息處理程序。
- conn：表示與對等端連接的WebSocketActor實例。
- token：用于處理任務(wù)取消，以實現(xiàn)安全關(guān)機。
handle_connection函數(shù)體邏輯如下：
1，提取套接字地址
// 提取套接字地址作為客戶端列表的鍵
let addr = match conn.ws_stream.get_ref() {
MaybeTlsStream::Plain(f) => f.peer_addr().unwrap(),
_ => {
panic!("tls is not supported yet");
}
};addr：對等端的socket地址。代碼從WebSocket流中提取對等端的地址。目前只支持非tls (plain)流。
2，設(shè)置消息通道
// 這個tx應(yīng)該在網(wǎng)絡(luò)狀態(tài)下共享
let (tx, mut rx) = unbounded_channel::<P2PInnerMessage>();
{
let mut list = state.addresses.lock().unwrap();
list.insert(addr, tx.clone());
}- tx和rx：用于在此函數(shù)和應(yīng)用程序的其他部分之間發(fā)送和接收消息(P2PInnerMessage)的無界通道。
- state.addresses：連接的對等端的地址存儲在一個共享的HashMap中，新對等端的地址和它的發(fā)送者(tx)被添加到列表中。
3，拆分WebSocket流
let (mut ws_tx, mut ws_rx) = conn.ws_stream.split();ws_tx和ws_rx：WebSocket流分為發(fā)送方(ws_tx)和接收方(ws_rx)，以異步方式處理傳入和傳出的消息。
4，循環(huán)消息處理
loop {
tokio::select! {
Some(msg) = ws_rx.next() => {
log::debug!("Received: {:?}", msg);
match msg {
Ok(msg) => {
if let Err(e) = state.master.lock().unwrap().send(P2PInnerMessage {
message: msg,
tx_handler: tx.clone(),
}) {
log::error!("Failed to send message to master: {:?}", e);
}
},
Err(e) => {
log::error!("Error receiving message or connection closed: {:?}", e);
break
}
}
}
Some(msg) = rx.recv() => {
log::debug!("Sending: {:?}", msg);
if let Err(e) = ws_tx.send(msg.message).await {
log::error!("Failed to send message on socket: {:?}", e);
}
}
_ = token.cancelled() => {
log::warn!("task cancelled");
break
}
}
}- 傳入消息(ws_rx)：循環(huán)使用tokio::select!等待多個異步事件。它首先檢查來自對等端的傳入消息(ws_rx.next())。如果成功接收到消息，則通過state.master將其轉(zhuǎn)發(fā)給主處理程序。這可以對消息進行集中處理或路由。如果發(fā)生錯誤(例如，連接關(guān)閉)，循環(huán)中斷，有效地終止連接。
- 傳出消息(rx)：循環(huán)還檢查從應(yīng)用程序的內(nèi)部通道(rx.recv())接收到的打算發(fā)送給對等端的消息。使用ws_tx.send(msg.message).await將消息發(fā)送到對等端。發(fā)送中的錯誤將被記錄。
- 取消令牌：如果取消令牌被觸發(fā)(Token .cancelled())，循環(huán)中斷，允許任務(wù)干凈地退出。
5，從列表中刪除對等端
{
// 從列表中刪除客戶端
let mut list = state.addresses.lock().unwrap();
list.remove(&addr);
}一旦循環(huán)退出(由于錯誤或取消)，對等體的地址將從連接地址列表中刪除，以確保狀態(tài)準確地反映當(dāng)前網(wǎng)絡(luò)連接。
handle_server_connection函數(shù)的操作邏輯與上述一樣，代碼如下：
async fn handle_server_connection(
state: Arc<P2PWebsocketNetwork>,
raw_stream: TcpStream,
addr: SocketAddr,
token: CancellationToken,
) {
let (tx, mut rx) = unbounded_channel::<P2PInnerMessage>();
{
let mut list = state.addresses.lock().unwrap();
list.insert(addr, tx.clone());
}
log::info!("Incoming TCP connection from: {}", addr);
let ws_stream = match tokio_tungstenite::accept_async(raw_stream).await {
Ok(ws) => ws,
Err(e) => {
log::error!("WebSocket handshake error: {:?}", e);
return;
}
};
log::info!("WebSocket connection established: {}", addr);
let (mut ws_tx, mut ws_rx) = ws_stream.split();
loop {
tokio::select! {
Some(msg) = ws_rx.next() => {
log::debug!("Received: {:?}", msg);
match msg {
Ok(msg) => {
if let Err(e) = state.master.lock().unwrap().send(P2PInnerMessage {
message: msg,
tx_handler: tx.clone(),
}) {
log::error!("Failed to send message to master: {:?}", e);
}
},
Err(e) => {
log::error!("Error receiving message or connection closed: {:?}", e);
break
}
}
}
Some(msg) = rx.recv() => {
log::debug!("Sending: {:?}", msg);
if let Err(e) = ws_tx.send(msg.message).await {
log::error!("Failed to send message on socket: {:?}", e);
}
}
_ = token.cancelled() => {
log::warn!("task cancelled");
break
}
}
}
{
// 從列表中刪除客戶端
let mut list = state.addresses.lock().unwrap();
list.remove(&addr);
}
}廣播消息
我們的廣播功能定期向所有連接的對等點發(fā)送消息，展示了P2P網(wǎng)絡(luò)傳播信息的能力。
async fn broadcast(
state: Arc<P2PWebsocketNetwork>,
tx: UnboundedSender<P2PInnerMessage>,
bind: String,
) {
log::debug!("Broadcast start");
// 廣播到已連接的客戶端
let list = state.addresses.lock().unwrap();
for (i, cl) in list.iter().enumerate() {
log::debug!("Broadcasting to {} ", cl.0);
if let Err(e) = cl.1.send(P2PInnerMessage {
message: tungstenite::protocol::Message::text(format!(
"Message to client {} from {}",
i, bind
)),
tx_handler: tx.clone(),
}) {
log::error!("Failed to send broadcast message: {:?}", e);
}
}
log::debug!("Broadcast end");
}main函數(shù)
#[tokio::main]
async fn main() {
let args = Args::parse();
env_logger::init();
let cancelation_token = CancellationToken::new();
let tracker = TaskTracker::new();
let (tx, mut rx) = unbounded_channel::<P2PInnerMessage>();
let network_state: Arc<P2PWebsocketNetwork> = Arc::new(P2PWebsocketNetwork {
addresses: Arc::new(Mutex::new(HashMap::new())),
master: Arc::new(Mutex::new(tx.clone())),
});
for url in &args.peers {
log::info!("connecting to {} ...", url);
if let Some(conn) = WebSocketActor::connect(url).await {
tracker.spawn(handle_connection(
network_state.clone(),
conn,
cancelation_token.clone(),
));
} else {
log::warn!("could not connect to server: {url}");
}
}
let listener = TcpListener::bind(&args.bind).await.expect("Failed to bind");
loop {
tokio::select! {
Ok((stream, addr)) = listener.accept() => {
tracker.spawn(handle_server_connection(
network_state.clone(),
stream, addr, cancelation_token.clone()));
}
Some(msg) = rx.recv() => {
log::debug!("consuming ->{msg:?}");
}
_ = tokio::signal::ctrl_c() => {
log::warn!("Received Ctrl+C, shutting down...");
tracker.close();
cancelation_token.cancel();
break
}
_ = tokio::time::sleep(tokio::time::Duration::from_secs(10)) => {
tracker.spawn(broadcast(network_state.clone(), tx.clone(), args.bind.clone()));
}
}
}
log::info!("waiting for all tasks");
tracker.wait().await;
log::debug!("tasks all are stoped");
}在這里 我們將通過處理Ctrl+C信號并相應(yīng)地取消任務(wù)來確保我們的網(wǎng)絡(luò)可以優(yōu)雅地關(guān)閉。
運行項目
在不同的終端中，使用這些參數(shù)按以下順序運行程序：
$ RUST_LOG=debug cargo run -- --bind localhost:8080 # 啟動第一個節(jié)點
$ RUST_LOG=debug cargo run -- --peers ws://localhost:8080 --bind localhost:8085 # 啟動第二個節(jié)點
$ RUST_LOG=debug cargo run -- --peers ws://localhost:8080,ws://localhost:8085 --bind localhost:8086 # 啟動第三個節(jié)點然后，將看到從每個對等點向所有其他對等點廣播消息。這是對所有人的廣播！
總結(jié)
這篇文章使用Rust構(gòu)建基于WebSocket的P2P網(wǎng)絡(luò)示例，它提供了一種強大而有效的方式來實現(xiàn)節(jié)點之間的實時通信。通過理解代碼的每個部分，可以擴展和定制這個示例以滿足特定需求，無論是分散的應(yīng)用程序，實時數(shù)據(jù)共享還是分布式計算任務(wù)。
責(zé)任編輯：武曉燕
來源：
coding到燈火闌珊
