如何使用 Rust 和 Tokio 构建高性能 WebSocket 聊天服务器

本文分享如何使用 Rust 和 Tokio 构建高性能 WebSocket 聊天服务器。我们将创建一个能处理多个并发连接、广播消息和管理用户状态的服务器。

项目设置

cargo new rust-chat-server
cd rust-chat-server

在Cargo.toml添加依赖项：

[package]
name = "rust-chat-server"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1.28", features = ["full"] }
tokio-tungstenite = "0.19"
futures-util = "0.3"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"

核心实现

创建 src/main.rs：

use futures_util::{SinkExt, StreamExt};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio_tungstenite::tungstenite::Message;

// 所有连接之间的共享状态
type Users = Arc<Mutex<HashMap<String, UserConnection>>>;

struct UserConnection {
    username: String,
    tx: futures_util::stream::SplitSink<
        tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>,
        Message,
    >,
}

#[derive(serde::Serialize, serde::Deserialize)]
struct ChatMessage {
    username: String,
    content: String,
    message_type: String,
}

#[tokio::main]
async fn main() {
    let addr = "127.0.0.1:8080";
    let users = Users::default();
    
    let listener = tokio::net::TcpListener::bind(&addr).await.unwrap();
    println!("WebSocket server listening on: {}", addr);

    while let Ok((stream, _)) = listener.accept().await {
        let users = Arc::clone(&users);
        tokio::spawn(handle_connection(stream, users));
    }
}

async fn handle_connection(stream: tokio::net::TcpStream, users: Users) {
    let ws_stream = tokio_tungstenite::accept_async(stream)
        .await
        .expect("Failed to accept");
    
    let (mut tx, mut rx) = ws_stream.split();
    
    // 等待用户名
    if let Some(Ok(msg)) = rx.next().await {
        if let Ok(chat_msg) = serde_json::from_str::<ChatMessage>(&msg.to_string()) {
            if chat_msg.message_type == "join" {
                let username = chat_msg.username;
                
                // 存储连接
                users.lock().await.insert(
                    username.clone(),
                    UserConnection {
                        username: username.clone(),
                        tx: tx.clone(),
                    },
                );

                // 广播加入消息
                broadcast_message(
                    &users,
                    &ChatMessage {
                        username: "System".to_string(),
                        content: format!("{} joined the chat", username),
                        message_type: "system".to_string(),
                    },
                )
                .await;

                // 处理消息
                while let Some(Ok(msg)) = rx.next().await {
                    if let Ok(chat_msg) = serde_json::from_str::<ChatMessage>(&msg.to_string()) {
                        broadcast_message(&users, &chat_msg).await;
                    }
                }

                //  处理断开连接
                users.lock().await.remove(&username);
                broadcast_message(
                    &users,
                    &ChatMessage {
                        username: "System".to_string(),
                        content: format!("{} left the chat", username),
                        message_type: "system".to_string(),
                    },
                )
                .await;
            }
        }
    }
}

async fn broadcast_message(users: &Users, msg: &ChatMessage) {
    let message = serde_json::to_string(&msg).unwrap();
    let users = users.lock().await;
    
    for user in users.values_mut() {
        if let Err(e) = user.tx.send(Message::Text(message.clone())).await {
            println!("Failed to send message to {}: {}", user.username, e);
        }
    }
}

HTML 客户端

创建一个简单的客户端来测试服务器（client.html）：

<!DOCTYPE html>
<html>
<head>
    <title>Rust Chat</title>
    <style>
        #messages { height: 300px; overflow-y: auto; border: 1px solid #ccc; }
        .system { color: #666; }
    </style>
</head>
<body>
    <div id="login">
        <input type="text" id="username" placeholder="Enter username">
        <button onclick="join()">Join Chat</button>
    </div>
    
    <div id="chat" style="display: none;">
        <div id="messages"></div>
        <input type="text" id="message" placeholder="Type a message">
        <button onclick="sendMessage()">Send</button>
    </div>

    <script>
        let ws;
        let username;

        function join() {
            username = document.getElementById('username').value;
            if (!username) return;

            ws = new WebSocket('ws://localhost:8080');
            
            ws.onopen = () => {
                ws.send(JSON.stringify({
                    username,
                    content: '',
                    message_type: 'join'
                }));
                document.getElementById('login').style.display = 'none';
                document.getElementById('chat').style.display = 'block';
            };

            ws.onmessage = (event) => {
                const msg = JSON.parse(event.data);
                const div = document.createElement('div');
                div.textContent = `${msg.username}: ${msg.content}`;
                if (msg.message_type === 'system') div.className = 'system';
                document.getElementById('messages').appendChild(div);
            };
        }

        function sendMessage() {
            const content = document.getElementById('message').value;
            if (!content) return;

            ws.send(JSON.stringify({
                username,
                content,
                message_type: 'chat'
            }));
            document.getElementById('message').value = '';
        }
    </script>
</body>
</html>