4. Warp
Warp 是一个构建在 Tokio
之上的 Web 框架,而且是一个非常好的框架。它与我们之前看到的其他框架非常不同。
Warp
与 Axum
有一些共同的特点:它构建在 Tokio
和 Hyper
之上,并利用了 Tower
中间件。然而,它在方法上有很大的不同。Warp
是建立在 Filter
trait 之上的。
在 Warp
中,我们构建一系列应用于传入请求的过滤器
,并将请求传递到管道直到达到末端。过滤器可以链接,它们可以组合。这使我们能够构建非常复杂的管道,但仍然易于理解。
Warp
也比 Axum
更接近 Tokio
生态系统,这意味着我们可能会在没有任何粘合特性的情况下处理更多 Tokio
结构和概念。
Warp
采用非常功能化的方法,如果这是我们的编程风格,我们将喜欢 Warp
的表达能力和可组合性。当我们查看 Warp
代码片段时,它通常读起来像正在发生的事情的故事,这在 Rust
中能够实现是有趣且令人惊讶的。
然而,随着这些不同的函数和过滤器被链接在一起,Warp
中的类型变得非常长且非常复杂,而且难以理解。错误消息也是如此,可能是难以理解的一大堆文本。
Warp
是一个很棒的框架。但是,它并不是最适合初学者的框架,也不是最流行的框架。这意味着我们可能在寻找帮助和资源方面会更加困难。但它非常适用于快速小型应用程序!
Warp 示例
来自其示例仓库的 WebSocket 聊天的 Warp 应用程序的简化示例:
// 定义了一个静态的原子 usize 计数器,用于为每个连接的用户分配唯一的用户ID。 static NEXT_USER_ID: AtomicUsize = AtomicUsize::new(1); // 当前连接用户的状态。 // 定义了一个类型别名 Users,它是一个原子引用计数的可读写锁的 HashMap,将用户ID映射到消息的发送器。 // Arc 是原子引用计数的智能指针,RwLock 是读写锁。 // - 键是其id // - 值是`warp::ws::Message`的发送器 type Users = Arc<RwLock<HashMap<usize, mpsc::UnboundedSender<Message>>>>; #[tokio::main] async fn main() { // 创建了一个 users 变量,用于存储连接的用户信息 let users = Users::default(); // 将其包装成 Warp 过滤器,以便在不同的路由中共享用户状态。 let users = warp::any().map(move || users.clone()); // chat 路由处理 WebSocket 握手 let chat = warp::path("chat") // `ws()`过滤器将准备WebSocket握手... .and(warp::ws()) .and(users) // 调用 user_connected 函数处理 WebSocket 连接。 .map(|ws: warp::ws::Ws, users| { // 如果握手成功,将调用我们的函数。 ws.on_upgrade(move |socket| user_connected(socket, users)) }); // 处理 HTTP GET 请求,返回一个包含聊天室链接的 HTML 页面 let index = warp::path::end().map(|| warp::reply::html(INDEX_HTML)); let routes = index.or(chat); warp::serve(routes).run(([127, 0, 0, 1], 3030)).await; } async fn user_connected(ws: WebSocket, users: Users) { // 使用计数器为此用户分配新的唯一ID。 let my_id = NEXT_USER_ID.fetch_add(1, Ordering::Relaxed); eprintln!("new chat user: {}", my_id); // 将套接字拆分为消息的发送器和接收器。 let (mut user_ws_tx, mut user_ws_rx) = ws.split(); // 创建一个新的消息通道 (mpsc::unbounded_channel) 用于将用户的消息广播给其他用户 let (tx, rx) = mpsc::unbounded_channel(); let mut rx = UnboundedReceiverStream::new(rx); tokio::task::spawn(async move { // 不断接收用户的消息。一旦用户断开连接,就会退出这个循环。 while let Some(message) = rx.next().await { user_ws_tx .send(message) .unwrap_or_else(|e| { eprintln!("websocket send error: {}", e); }) .await; } }); //将发送器保存在我们的已连接用户列表中。 users.write().await.insert(my_id, tx); // 返回一个基本上是管理此特定用户连接的状态机的“Future”。 // 每当用户发送消息时,将其广播给 // 所有其他用户... while let Some(result) = user_ws_rx.next().await { let msg = match result { Ok(msg) => msg, Err(e) => { eprintln!("websocket error(uid={}): {}", my_id, e); break; } }; user_message(my_id, msg, &users).await; } // 只要用户保持连接,user_ws_rx流就会继续处理。一旦他们断开连接,那么... user_disconnected(my_id, &users).await; } // 处理用户发送的消息。它跳过非文本消息,将文本消息格式化为 <User#ID>: Message,然后将其广播给所有其他用户。 async fn user_message(my_id: usize, msg: Message, users: &Users) { // 跳过任何非文本消息... let msg = if let Ok(s) = msg.to_str() { s } else { return; }; let new_msg = format!("<User#{}>: {}", my_id, msg); // 来自此用户的新消息,将其发送给所有其他用户(除了相同的uid)... for (&uid, tx) in users.read().await.iter() { if my_id != uid { if let Err(_disconnected) = tx.send(Message::text(new_msg.clone())) { // 发送器已断开连接,我们的`user_disconnected`代码 // 应该在另一个任务中执行,这里没有更多的事情要做。 } } } } async fn user_disconnected(my_id: usize, users: &Users) { eprintln!("good bye user: {}", my_id); // 流关闭,因此从用户列表中删除 users.write().await.remove(&my_id); }
Warp 特点
- 函数式方法。
- 良好的表达能力。
- 通过接近
Tokio
、Tower
和Hyper
构建强大的生态系统。 - 不适合初学者的框架
5. Tide
Tide 是一个建立在 async-std
运行时之上的极简主义 Web 框架。极简主义的方法意味着我们得到了一个非常小的 API 表面。Tide
中的处理函数是 async fn
,接受一个 Request
并返回一个 Response
的 tide::Result
。提取数据或发送正确的响应格式由我们自行完成。
虽然这可能对我们来说是更多的工作,但也更直接,意味着我们完全掌控正在发生的事情。在某些情况下,能够离 HTTP 请求和响应如此近是一种愉悦,使事情变得更容易。
Tide
的中间件方法与我们从 Tower
中了解的类似,但 Tide
公开了 async trait
crate,使实现变得更加容易。
Tide 示例
来自其示例仓库的用户会话示例:
// async-std crate 提供的异步 main 函数。它返回一个 Result,表示可能的错误。 #[async_std::main] async fn main() -> Result<(), std::io::Error> { // 使用 femme crate 启用颜色日志。这是一个美观的日志记录库,可以使日志输出更易读。 femme::start(); // 创建一个 Tide 应用程序实例 let mut app = tide::new(); // 添加一个日志中间件,用于记录请求和响应的日志信息。 app.with(tide::log::LogMiddleware::new()); // 添加一个会话中间件,用于处理会话数据。这里使用内存存储,并提供一个密钥(TIDE_SECRET),用于加密和验证会话数据。 app.with(tide::sessions::SessionMiddleware::new( tide::sessions::MemoryStore::new(), std::env::var("TIDE_SECRET") .expect( "Please provide a TIDE_SECRET value of at \ least 32 bytes in order to run this example", ) .as_bytes(), )); // 添加一个 Before 中间件,它在处理请求之前执行。在这里,它用于增加访问计数,存储在会话中。 app.with(tide::utils::Before( |mut request: tide::Request<()>| async move { let session = request.session_mut(); let visits: usize = session.get("visits").unwrap_or_default(); session.insert("visits", visits + 1).unwrap(); request }, )); // 定义了一个处理根路径的GET请求的路由。这个路由通过 async move 来处理请求,获取会话中的访问计数,并返回一个包含访问次数的字符串。 app.at("/").get(|req: tide::Request<()>| async move { let visits: usize = req.session().get("visits").unwrap(); Ok(format!("you have visited this website {} times", visits)) }); // 定义了一个处理 "/reset" 路径的GET请求的路由。这个路由通过 async move 处理请求,将会话数据清除,然后重定向到根路径 app.at("/reset") .get(|mut req: tide::Request<()>| async move { req.session_mut().destroy(); Ok(tide::Redirect::new("/")) }); // 启动应用程序并监听在 "127.0.0.1:8080" 地址上。使用 await? 处理可能的启动错误。 app.listen("127.0.0.1:8080").await?; Ok(()) }
Tide 简要概述
- 极简主义方法。
- 使用
async-std
运行时。 - 简单的处理函数。
- 异步特性的试验场。
6. Poem
Poem 声称自己是一个功能齐全但易于使用的 Web 框架。乍一看,它的使用方式与 Axum
非常相似,唯一的区别是它需要使用相应的宏标记处理程序函数。它还建立在 Tokio
和 Hyper
之上,完全兼容 Tower
中间件,同时仍然暴露自己的中间件特性。
Poem
的中间件特性也非常简单易用。我们可以直接为所有或特定的 Endpoint
(Poem 表达一切都可以处理 HTTP 请求的方式)实现该特性,或者只需编写一个接受 Endpoint
作为参数的异步函数。
Poem
不仅与更广泛的生态系统中的许多功能兼容,而且还具有丰富的功能,包括对 OpenAPI
和 Swagger
文档的全面支持。它不仅限于基于 HTTP
的 Web 服务,还可以用于基于 Tonic
的 gRPC
服务,甚至在 Lambda
函数中使用,而无需切换框架。添加对 OpenTelemetry
、Redis
、Prometheus
等的支持,我们就可以勾选所有现代企业级应用程序 Web 框架的所有框。
Poem
仍然处于 0.x
版本,但如果保持势头并交付出色的 1.0 版本,这将是一个值得关注的框架!
Poem 示例
来自其示例仓库的 WebSocket 聊天的缩写版本:
// 注解表示这是一个处理器函数,用于处理 WebSocket 请求 #[handler] fn ws( // 提取了 WebSocket 路径中的名字参数 Path(name): Path<String>, // WebSocket 对象,表示与客户端的连接 ws: WebSocket, // 是一个数据提取器,用于获取广播通道的发送器。 sender: Data<&tokio::sync::broadcast::Sender<String>>, ) -> impl IntoResponse { // 克隆了广播通道的发送器 sender。 let sender = sender.clone(); // 它订阅了广播通道,创建了一个接收器 receiver let mut receiver = sender.subscribe(); // 处理 WebSocket 连接升级 ws.on_upgrade(move |socket| async move { // 将连接的读写部分拆分为 sink 和 stream let (mut sink, mut stream) = socket.split(); // 从 WebSocket 客户端接收消息 // 如果是文本消息,则将其格式化为 {name}: {text} 的形式,并通过广播通道发送。 // 如果发送失败(例如,通道关闭),则任务终止。 tokio::spawn(async move { while let Some(Ok(msg)) = stream.next().await { if let Message::Text(text) = msg { if sender.send(format!("{name}: {text}")).is_err() { break; } } } }); // 从广播通道接收消息,并将其发送到 WebSocket 客户端 tokio::spawn(async move { while let Ok(msg) = receiver.recv().await { if sink.send(Message::Text(msg)).await.is_err() { break; } } }); }) } #[tokio::main] async fn main() -> Result<(), std::io::Error> { // 使用 tide::Route 创建了一个路由,其中包括两个路径: // - / 路径处理 HTTP GET 请求,调用 index 函数。 // - /ws/:name 路径处理 WebSocket 请求,调用 ws 函数。 let app = Route::new().at("/", get(index)).at( "/ws/:name", // 通过 tokio::sync::broadcast::channel 创建一个广播通道; // 并通过 tokio::sync::broadcast::channel::<String>(32).0 // 获取其发送器,将其作为数据传递给 ws 处理函数 get(ws.data(tokio::sync::broadcast::channel::<String>(32).0)), ); // 创建了一个服务器实例 Server::new(TcpListener::bind("127.0.0.1:3000")) // 启动服务器,并等待其完成运行。 .run(app) .await }
Poem 简要概述
- 丰富的功能集。
- 与 Tokio 生态系统兼容。
- 易于使用。
- 适用于 gRPC 和 Lambda。
后记
正如我们所见,Rust Web
框架的世界非常多样化。没有一种解决方案适用于所有情况,我们需要选择最符合我们需求的框架。如果我们刚刚开始,我建议我们选择 Actix
或 Axum
,因为它们是最适合初学者的框架,而且它们有着出色的文档。
分享是一种态度。
全文完,既然看到这里了,如果觉得不错,随手点个赞和“在看”吧。