如何在Actix Web WebSocket Interval中调用异步函数？

我尝试在Actix Web的WebSocket定时任务（Interval）中运行异步代码，但不知道该怎么实现。我查过相关问题，但因为我的问题在设计和执行上略有不同，不确定是否该用同样的解决方案。以下是我的代码：

struct WebSocket {
    data: Data<State>,
    user: ID,
}

const POLL_INTERVAL: Duration = Duration::from_millis(5);

impl Actor for WebSocket {
    type Context = WebsocketContext<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        ctx.run_interval(POLL_INTERVAL, |myself, ctx| {
            // let fut = ws_poll(myself.user, &myself.data.event_client);

            // fut.into_actor(myself).spawn(ctx);
            // I want to do async stuff in here
            // my_async_fn().await;
        });
    }
}

impl StreamHandler<Result<Message, ProtocolError>> for WebSocket {
    fn handle(&mut self, item: Result<Message, ProtocolError>, ctx: &mut Self::Context) {
        match item.unwrap() {
            Message::Text(txt) => match serde_json::from_str::<WebSocketMessage>(&txt.to_string()).unwrap() {
                WebSocketMessage::SendChatMessage { chat, message } => {}
                WebSocketMessage::ReadChat(chat) => {}
                WebSocketMessage::Typing(chat) => {}
                _ => panic!()
            }
            Message::Ping(msg) => ctx.pong(msg.as_ref()),
            Message::Close(_) => todo!(),
            _ => panic!(),
        }
    }
}

#[get("/ws")]
async fn ws_index(data: Data<State>, user: AuthedUser, r: HttpRequest, stream: Payload) -> WebResult<HttpResponse> {
    ws::start(WebSocket { data, user: user.id }, &r, stream)
}

pub fn config(cfg: &mut web::ServiceConfig) {
    cfg.service(ws_index);
}

解决方案 #

Actix Web的WebsocketContext::run_interval接收同步闭包，但可以通过将异步任务转换为Actor兼容的Future来执行异步逻辑，具体步骤如下：

1. 包装异步任务为ActorFuture：用into_actor方法把普通异步函数的Future转换为绑定当前WebSocket Actor实例的ActorFuture，这样能在任务中访问Actor的状态（比如myself.user和myself.data）。
2. 提交任务到上下文：调用spawn方法把转换后的Future提交到WebSocketContext中执行。
3. 添加错误处理：避免直接unwrap，用match或if let处理异步结果，防止panic导致连接中断。

修改后的started方法示例：

fn started(&mut self, ctx: &mut Self::Context) {
    ctx.run_interval(POLL_INTERVAL, |myself, ctx| {
        // 定义异步任务
        let fut = async move {
            // 执行你的异步逻辑，比如调用ws_poll
            let result = ws_poll(myself.user, &myself.data.event_client).await;
            
            // 处理异步结果，比如向客户端发送消息
            match result {
                Ok(poll_data) => {
                    if let Ok(json_msg) = serde_json::to_string(&poll_data) {
                        ctx.text(json_msg);
                    }
                }
                Err(e) => {
                    eprintln!("Poll task failed: {}", e);
                    // 可选：错误时关闭连接或发送错误提示
                    // ctx.text("Poll error occurred");
                }
            }
        };

        // 将异步任务转换为ActorFuture并提交执行
        fut.into_actor(myself)
            .map(|_, _, _| ())
            .drop_err()
            .spawn(ctx);
    });
}

关键说明 #

• into_actor(myself)：把异步Future和当前Actor实例绑定，允许在Future中访问Actor的状态。
• map(|_, _, _| ())：处理Future完成后的收尾逻辑，这里仅标记任务结束。
• drop_err()：忽略异步任务中的错误，也可以用map_err自定义错误处理逻辑。
• 注意当前设置的5毫秒间隔过短，可能引发性能问题，建议根据实际业务需求调整。

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