Rust WebSocket流处理库ws_stream_tungstenite的使用指南

ws_stream_tungstenite是一个基于tungstenite的Rust库，提供了更高级的WebSocket流处理功能，特别适合需要高效处理WebSocket通信和数据流的场景。

主要特性

• 基于成熟的tungstenite WebSocket实现
• 提供流式接口处理WebSocket消息
• 支持异步/等待(async/await)模式
• 简化了WebSocket通信的复杂性
• 良好的错误处理和连接管理

安装

在Cargo.toml中添加依赖：

[dependencies]
ws_stream_tungstenite = "0.5"
tokio = { version = "1.0", features = ["full"] }

基本使用方法

客户端连接示例

use ws_stream_tungstenite::*;
use futures::prelude::*;
use tokio::net::TcpStream;

#[tokio::main]
async fn main() {
    let url = "ws://echo.websocket.org";
    
    // 建立TCP连接
    let tcp = TcpStream::connect("echo.websocket.org:80").await.unwrap();
    
    // 创建WebSocket连接
    let (ws_stream, _) = tokio_tungstenite::connect_async(url)
        .await
        .unwrap();
    
    // 转换为WsStream
    let mut ws_stream = WsStream::new(ws_stream);
    
    // 发送消息
    ws_stream.send("Hello, WebSocket!".into()).await.unwrap();
    
    // 接收消息
    while let Some(msg) = ws_stream.next().await {
        match msg {
            Ok(msg) => println!("Received: {:?}", msg),
            Err(e) => eprintln!("Error: {}", e),
        }
    }
}

服务器端示例

use ws_stream_tungstenite::*;
use futures::prelude::*;
use tokio::net::{TcpListener, TcpStream};

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:8080").await.unwrap();
    
    while let Ok((tcp_stream, _)) = listener.accept().await {
        tokio::spawn(handle_connection(tcp_stream));
    }
}

async fn handle_connection(tcp_stream: TcpStream) {
    // 升级为WebSocket连接
    let ws_stream = tokio_tungstenite::accept_async(tcp_stream)
        .await
        .unwrap();
    
    let mut ws_stream = WsStream::new(ws_stream);
    
    // 处理消息
    while let Some(msg) = ws_stream.next().await {
        match msg {
            Ok(msg) => {
                println!("Received: {:?}", msg);
                // 回显消息
                ws_stream.send(msg).await.unwrap();
            }
            Err(e) => {
                eprintln!("Error: {}", e);
                break;
            }
        }
    }
}

高级用法

处理二进制数据

#[tokio::main]
async fn main() {
    let (mut ws_stream, _) = WsStream::connect("ws://example.com/ws")
        .await
        .unwrap();
    
    // 发送二进制数据
    let binary_data = vec![0x01, 0x02, 0x03, 0x04];
    ws_stream.send(binary_data.into()).await.unwrap();
    
    // 接收二进制数据
    while let Some(msg) = ws_stream.next().await {
        if let Ok(Message::Binary(data)) = msg {
            println!("Received binary data: {:?}", data);
        }
    }
}

使用Sink特性批量发送消息

use futures::SinkExt;

#[tokio::main]
async fn main() {
    let (mut ws_stream, _) = WsStream::connect("ws://example.com/ws")
        .await
        .unwrap();
    
    let messages = vec![
        Message::text("Message 1"),
        Message::text("Message 2"),
        Message::text("Message 3"),
    ];
    
    let mut sink = ws_stream.sink();
    sink.send_all(&mut futures::stream::iter(messages))
        .await
        .unwrap();
}

错误处理

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (mut ws_stream, _) = WsStream::connect("ws://example.com/ws")
        .await?;
    
    ws_stream.send("Ping".into()).await?;
    
    match ws_stream.next().await {
        Some(Ok(msg)) => println!("Received: {:?}", msg),
        Some extreme(Err(e)) => eprintln!("WebSocket error: {}", e),
        None => println!("Connection closed"),
    }
    
    Ok(())
}

性能提示

1. 对于高频消息场景，考虑使用Message::Binary而不是Message::Text，因为二进制数据通常处理效率更高
2. 批量处理消息时使用Sink特性可以提高性能
3. 合理设置TCP缓冲区大小可以改善吞吐量

完整示例demo

下面是一个完整的客户端和服务器端交互的WebSocket示例：

服务器端完整代码

use ws_stream_tungstenite::*;
use futures::prelude::*;
use tokio::net::{TcpListener, TcpStream};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 绑定到本地8080端口
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("WebSocket服务器启动，监听端口8080...");

    while let Ok((tcp_stream, _)) = listener.accept().await {
        tokio::spawn(async move {
            if let Err(e) = handle_connection(tcp_stream).await {
                eprintln!("连接处理错误: {}", e);
            }
        });
    }
    
    Ok(())
}

async fn handle_connection(tcp_stream: TcpStream) -> Result<(), Box<dyn std::error::Error>> {
    // 升级为WebSocket连接
    let ws_stream = tokio_tungstenite::accept_async(tcp_stream).await?;
    let mut ws_stream = WsStream::new(ws_stream);
    
    println!("新客户端连接");

    // 处理消息循环
    while let Some(msg) = ws_stream.next().await {
        match msg {
            Ok(msg) => {
                println!("收到消息: {:?}", msg);
                // 回显消息给客户端
                ws_stream.send(msg).await?;
            }
            Err(e) => {
                eprintln!("错误: {}", e);
                break;
            }
        }
    }
    
    println!("客户端断开连接");
    Ok(())
}

客户端完整代码

use ws_stream_tungstenite::*;
use futures::prelude::*;
use tokio::net::TcpStream;
use futures::SinkExt;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 连接WebSocket服务器
    let url = "ws://127.0.0.1:8080";
    let tcp = TcpStream::connect("127.0.0.1:8080").await?;
    
    // 创建WebSocket连接
    let (ws_stream, _) = tokio_tungstenite::connect_async(url).await?;
    let mut ws_stream = WsStream::new(ws_stream);
    
    println!("成功连接到服务器");

    // 创建Sink用于批量发送
    let mut sink = ws_stream.sink();
    
    // 批量发送文本消息
    let text_messages = vec![
        Message::text("第一条消息"),
        Message::text("第二条消息"),
        Message::text("第三条消息"),
    ];
    sink.send_all(&mut futures::stream::iter(text_messages)).await?;

    // 发送二进制消息
    let binary_data = vec![0x01, 0x02, 0x03, 0x04];
    ws_stream.send(binary_data.into()).await?;

    // 接收服务器响应
    while let Some(msg) = ws_stream.next().await {
        match msg {
            Ok(msg) => println!("收到服务器响应: {:?}", msg),
            Err(e) => {
                eprintln!("错误: {}", e);
                break;
            }
        }
    }
    
    Ok(())
}

这个完整示例展示了：

1. 服务器端启动WebSocket服务并回显接收到的消息
2. 客户端连接服务器并发送文本和二进制消息
3. 使用Sink特性批量发送消息
4. 完整的错误处理机制

您可以将服务器端和客户端代码分别保存为两个文件，先运行服务器端，再运行客户端进行测试。