前面十几篇把 Rust 的核心机制拆开讲了一遍所有权管内存trait 定义行为错误当成值来传并发在编译期被拦住。但零件认全了不等于会组装。这一篇换一种方式——不再逐条介绍概念而是搭出一个真能跑起来的小项目让那些机制在同一个工程里各归其位。项目是一个RTMP 流媒体服务器。功能很简单接收推流端比如 FFmpeg送来的音视频数据按流 key 分发给所有拉流端比如播放器并在连接和推流时做鉴权。跑起来之后你可以一边用 FFmpeg 推、一边用播放器拉同时看到服务器打出的连接、推流、拉流、结束的完整日志。用到的库标准库std自带不用写进依赖std::collections的HashSet和HashMap前者记录谁在推流后者接收元数据回调传来的键值对std::sync的Mutex和AtomicUsize让多份连接共享的状态能被安全修改std::net的地址类型解析0.0.0.0:1935第三方库写进Cargo.tomlrtmp-rsRTMP 协议的实现提供服务器骨架RtmpServer和一份回调契约RtmpHandler。协议细节由它处理你只写业务规则tokio异步运行时。服务器要同时应付很多连接靠它驱动tracing和tracing-subscriber日志。前者是打日志的宏后者负责输出到终端并读RUST_LOG控制级别dotenvy读取.env文件让日志级别不用改代码就能调publishstream_key live_test转发帧转发帧ffmpeg推流端RTMP 服务器:1935播放器 A拉流 live_test播放器 B拉流 live_test效果项目结构cargonew rtmp-servercdrtmp-servercargo new只会生成Cargo.toml、src/main.rs和一个.gitignore。我们要再建两个文件最终目录结构是这样rtmp-server/ ├── Cargo.toml ← 改写依赖 ├── .env ← 新建 └── src/ ├── main.rs ← 改写 └── handler.rs ← 新建入口依赖就是rtmp-rs、tokio、tracing加一个读.env的dotenvy。入口只有十几行modhandler;usecrate::handler::MyHandler;usertmp_rs::{RtmpServer,ServerConfig};#[tokio::main]asyncfnmain()-Result(),Boxdynstd::error::Error{letconfigServerConfig::default().bind(0.0.0.0:1935.parse()?);letserverRtmpServer::new(config,MyHandler::new());server.run().await?;Ok(())}mod第一行的mod handler;要特别说明它不是“引入”某个文件而是声明“这里有一个叫 handler 的模块”。编译器据此去找src/handler.rs。一个.rs文件只有被某处的mod声明到才会被编译进去——单纯把文件放进src/目录什么也不会发生。crateuse crate::handler::MyHandler;里的crate指本 crate 的根也就是main.rs。pub由于MyHandler在handler.rs里标了pub这行才成立。env顺序上只有一处要求dotenvy::dotenv()必须在日志初始化之前因为EnvFilter::try_from_default_env()读的是进程环境变量。所有权注意MyHandler::new()是按值传进RtmpServer::new的而run只需要self不是mut self。也就是说run拿到你的 handler 之后要自己想办法把它分发给每一个连接任务——它不能独占因为你已经把所有权交出去了。这正是后面所有约束的起点。「把帧转发给所有订阅者」这件事rtmp-rs内部已经做完了代码里只需处理事件回调。约束一self决定了字段用什么类型pubstructMyHandler{/// 正在推流的 stream keypublishing:MutexHashSetString,/// 当前连接数connections:AtomicUsize,}两个字段两种工具。它们要解决的是同一个问题只有self怎么改数据。RtmpHandler的每个回调第一个参数都是self。从共享引用里拿不到可变访问所以self.connections 1这种写法直接编译不过。绕开的办法有两个原子类型AtomicUsize的fetch_add和fetch_sub本身就是以共享引用接收self的不需要可变借用。单个计数的增减就能这样完成。锁HashSet的插入是「读—改—写」三步还要处理哈希冲突没有原子版本可用只能套Mutex把这段操作圈成临界区。这里有个容易忽略的坑RefCell看起来也是为「共享引用下修改」准备的但它用不了。RefCell不是Sync多个连接同时访问会编译失败——它只能用于单线程内。顺带解释drop(set)那一行letmutsetself.publishing.lock().unwrap();letremovedset.remove(stream_key);letonlineset.len();drop(set);// 临界区到此为止日志在锁外打临界区只覆盖remove和len()。打日志不需要持锁提前把守卫放掉别的线程就能立刻进来改集合。反过来如果把日志格式化留在锁内别的连接就得等着日志打完才能动这个集合。约束二为什么这里不需要Arc我最早写这个项目时字段上是套了Arc的ArcMutexHashSetString。这是从「多个任务共享状态」这个直觉直接推出来的写法用tokio::spawn用过一次就会形成肌肉记忆。但在这里它是多余的。判断依据是那条契约。RtmpHandler要求实现者是Send Sync static而库要把同一份实例分发给每一个连接任务——分发这件事是库自己做的它内部必然已经拿了Arc或等价的引用计数来持有你的 handler。而你需要保证的只有一件事让ArcH是Send Sync。这要求H本身是Send Sync。也就是说你该做的是让字段满足这两条标记 trait而不是自己再套一层Arc去提供共享——共享已经有人做了。什么时候才真的需要自己加Arc当你要把这份状态交给库之外的任务时。比如加一个定时任务定期清理超时未结束的推流记录那个任务需要自己持有一个句柄letpublishingArc::new(Mutex::new(HashSet::new()));letfor_cleanupArc::clone(publishing);tokio::spawn(asyncmove{// 定期检查 for_cleanup});这时候是两个Arc各司其职库持有它自己那份清理任务持有你新克隆的这一份两边指向同一块数据。如果不需要库之外的人访问那就一层都不用加。约束三锁不能跨awaitasync fn会被编译成一个状态机跨越await点还活着的值会成为这个 future 的字段。这条规则带来一个直接后果如果MutexGuard跨过了await它就成了 future 的一部分而标准库的MutexGuard不是Send于是整个 future 也不再是Send多线程执行器没法调度它——编译直接失败。这些回调体里一个await都没有所以锁天然不会跨界。但这不代表可以随便写。一旦某个回调里加了异步调用就得小心守卫的存活范围// 反面写法守卫活过了 awaitletguardself.state.lock().unwrap();some_async_call().await;// guard 跨过挂起点future 不再是 Send正确做法是用一对花括号把守卫圈住让它在await之前就销毁。这也是为什么很多 Rust 项目里能频繁看到看似多余的块作用域——它们不是装饰是在给守卫划边界。AuthResult拒绝的决定和拒绝理由是同一个值流 key 校验失败时回调这样返回asyncfnon_fc_publish(self,ctx:SessionContext,stream_key:str)-AuthResult{matchself.check_stream_key(stream_key){Ok(())AuthResult::Accept,Err(reason){warn!(session_idctx.session_id,stream_key,reason,FCPublish 拒绝);AuthResult::Reject(reason.to_string())}}}AuthResult是个普通枚举三个变体Accept、Reject(String)、Redirect { url: String }。三种情况携带的数据完全不同——一个什么都没有一个带理由一个带命名的url字段。值得看的是Reject(String)这个设计。在异常式的模型里「拒绝」是一次非局部的控制流跳转「理由」是挂在异常对象上的附带信息两者走在不同的通道上。而在 Rust 里它们是同一个值——你不可能返回一个「拒绝」却忘了带上理由因为不构造那个String就没有Reject可返回。Redirect同理用命名的url字段而不是裸字符串是为了让match分支读起来知道自己在取什么。match的穷尽性检查在这里也发挥作用三个变体要么全覆盖要么显式写_表示「剩下的我不管」。编译器不会让你默默漏掉一种处理方式。完整代码在 https://github.com/RambleTechnology/rust-rtmp-server可以直接cargo run起来推拉流试试。参考来源RtmpHandler trait 文档 — 全部回调签名、默认实现与触发时机RtmpServer 文档 — new、run(self)、bind_addr 的签名handler.rs 源码 — AuthResult 三个变体与各回调签名项目完整代码