Rust中基于Actix-Web通过WebSocket修改共享可变状态的问题

核心思路是在WebSocket会话初始化阶段捕获actix_web::web::Data，并将其存入会话的自定义状态结构体中，这样StreamHandler的handle方法就能通过self访问到共享状态，无需修改trait的方法签名。

步骤1：定义线程安全的共享状态 #

首先需要确保共享状态满足Send + Sync（线程安全），通常用Mutex或RwLock包装内部可变数据：

use actix_web::{web, Error, HttpResponse};
use actix_web_actors::ws;
use std::sync::Mutex;

// 全局共享的应用状态
#[derive(Clone)]
struct AppState {
    counter: Mutex<i32>, // 用Mutex保证多线程下的安全可变访问
}

步骤2：定义WebSocket会话状态 #

创建自定义的WebSocket会话结构体，将Data<AppState>作为成员变量保存：

// WebSocket会话状态，持有共享应用状态的引用
struct WsSession {
    app_state: web::Data<AppState>,
}

impl WsSession {
    // 构造函数：接收共享状态并初始化会话
    fn new(app_state: web::Data<AppState>) -> Self {
        WsSession { app_state }
    }
}

步骤3：实现Actor和StreamHandler trait #

在实现StreamHandler的handle方法时，直接通过self.app_state访问并修改共享状态：

impl actix::Actor for WsSession {
    type Context = ws::WebsocketContext<Self>;
}

impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsSession {
    fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
        match msg {
            Ok(ws::Message::Text(text)) => {
                // 示例：根据WebSocket消息修改共享计数器
                match text.as_str() {
                    "increment" => {
                        let mut counter = self.app_state.counter.lock().unwrap();
                        *counter += 1;
                        ctx.text(format!("Counter updated to: {}", counter));
                    }
                    "decrement" => {
                        let mut counter = self.app_state.counter.lock().unwrap();
                        *counter -= 1;
                        ctx.text(format!("Counter updated to: {}", counter));
                    }
                    _ => ctx.text("Unknown command"),
                }
            }
            Ok(ws::Message::Close(reason)) => {
                ctx.close(reason);
                ctx.stop();
            }
            _ => ctx.stop(),
        }
    }
}

步骤4：在路由中初始化WebSocket会话 #

在WebSocket路由处理函数中，从请求提取Data<AppState>，传递给WsSession并启动会话：

async fn ws_route(
    req: actix_web::HttpRequest,
    stream: web::Payload,
    app_state: web::Data<AppState>,
) -> Result<HttpResponse, Error> {
    let session = WsSession::new(app_state);
    let resp = ws::start(session, &req, stream)?;
    Ok(resp)
}

// 启动服务器并注册路由与共享状态
#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let app_state = web::Data::new(AppState {
        counter: Mutex::new(0),
    });

    actix_web::HttpServer::new(move || {
        actix_web::App::new()
            .app_data(app_state.clone())
            .route("/ws", web::get().to(ws_route))
    })
    .bind(("127.0.0.1", 8080))?
    .run()
    .await
}

关键说明 #

• web::Data是线程安全的智能指针，克隆操作仅增加引用计数，不会复制内部数据。
• 共享状态内部必须用同步原语（如Mutex、RwLock）包裹，避免多WebSocket会话并发访问时出现数据竞争。
• 通过将Data存入会话状态，StreamHandler的handle方法完全符合trait签名要求，无需修改框架代码。

内容的提问来源于stack exchange，提问作者LBlend