简介这是一份基于Netty开发的RTMP服务器端实现面向Java网络编程与流媒体方向的中高级开发者适合用来自建RTMP推流、拉流服务并深入理解协议解析与高并发连接处理。项目核心利用Netty的ChannelHandler解析RTMP数据包结合EventLoopGroup完成线程调度配合FFmpeg即可实现推流验证覆盖从服务端搭建到音视频数据分发的完整链路。资源包约59KB共50个文件其中包含16个Java源文件、19个class编译产物、9个XML配置以及工程模块文件和说明文档源码、配置与构建信息一应俱全便于直接导入IDE研读或二次改造。已有767人学习浏览。通过这份资料开发者可以掌握RTMP握手、消息分块、推流与拉流交互逻辑并结合目录结构与Netty事件模型理解真实项目的组织方式。1. 用 Netty 写 RTMP 推流服务器先弄清 rtmpServer-master 这个工程能给你什么你拿到一个叫 rtmpServer-master 的工程第一反应大概是终于不用自己从头抠 RTMP 协议了。这个包确实是一套基于 Netty 开发 RTMP 推流服务端的工程骨架覆盖了握手、chunk stream 解码、AMF0 命令解析和媒体数据转发这条最核心链路。它能解决的是「让 ffmpeg / OBS 把流推进来服务端稳定收下来」这件事至于转播、录制、鉴权通常还得自己在上面加。Netty 本身不懂 RTMP它只负责把 TCP 字节流切成事件喂给你剩下的协议状态机全部要你在 pipeline 里手工实现。刚开始接触服务端直播或者想把 RTMP 收流做成内部服务的工程师拿它做起点比从零开始省一个量级的精力但想直接丢到生产环境接千路推流你必须清楚它的边界在哪里。2. rtmpServer-master 要改得动先搞懂 RTMP 协议和 Netty pipeline 的映射2.1 RTMP 协议不是平铺的握手、块流、消息三层分别对应哪些 Netty 职责RTMP 跑在 TCP 上本质是一条没有帧边界的字节流所以服务端第一件事是状态机。连接建立后客户端先发 C01 字节版本号和 C11536 字节随机数据服务端回 S0S1S2客户端再回 C2这才算握手完成。握手之后的字节流被切成 chunk块每个块有 basic header、message header 和扩展时间戳再拼回一个完整的 message。message 里可能是 AMF0 编码的命令也可能直接是音频或视频数据。这块映射到 Netty 时我一般把职责拆成三个握手解码器只处理 C0/C1/C2chunk 解码器负责把块拼成 message处理粘包和拆包message handler 负责调用 AMF0 解析和业务状态机。千万不要在同一个 handler 里又拼 chunk 又解析命令否则很快会被边界条件搞到翻车。rtmpServer-master 这类工程的价值就是把这个分层已经摆好了你要做的是看懂每一层消费的是什么而不是把所有逻辑堆在一起改。2.2 Netty pipeline 怎么编排先握手再 chunk最后业务处理常见做法是让握手解码器放在最前面握手完成后把剩余字节交给后面的 chunk 解码器。不要一上来就把 chunk 解码器也挂上因为握手数据本身不是合法 chunk后面的解码器如果不知道当前状态会把 C0 当 basic header 解析直接错位。一个最小可用的 pipeline 长这样public final class RtmpServerInitializer extends ChannelInitializerSocketChannel { Override protected void initChannel(SocketChannel ch) { ChannelPipeline pipeline ch.pipeline(); // 读超时 60 秒防止推流端半开连接一直占着资源 pipeline.addLast(idle, new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS)); // 握手解码器握手完成后会把剩余数据往下传 pipeline.addLast(handshake, new RtmpHandshakeDecoder()); // chunk 解码器负责把 RTMP chunk 拼成完整消息 pipeline.addLast(chunk, new RtmpChunkDecoder()); // AMF0 命令解析把命令、事务 ID、参数从 payload 里读出来 pipeline.addLast(amf, new RtmpMessageDecoder()); // 业务处理维护连接、房间、流状态 pipeline.addLast(handler, new RtmpMessageHandler()); } }这里有几个参数值得注意IdleStateHandler(60, 0, 0, TimeUnit.SECONDS)表示 60 秒内没有读事件就触发userEventTriggered我们在这个回调里主动关闭连接用来兜底拔网线和进程崩溃造成的半开连接。handshake放在chunk之前chunk又放在amf之前顺序不能乱如果 C2 和后面的 connect 命令在同一批字节里到达handshake必须把 C2 消费掉、把剩余部分作为输出传给chunk否则客户端会觉得服务端没有响应。2.3 chunk 解码器为什么必须单独成一类消息边界、流 ID 和 chunk size 都在这里管RTMP 默认 chunk size 是 128 字节一个 50KB 的视频关键帧会被拆成 400 多个 chunk 依次到达。如果解码器不按 message length 做累积每收到一个 chunk 就往外抛消息后面的 AMF 和媒体处理会全乱。chunk 解码器要做的是从 basic header 里判断当前 chunk 的 fmt 和 csid算出本次 chunk 的头部长度再根据 fmt 判断有没有 message header然后按 message header 里的 message length 等完整 payload等齐了再输出一个完整的RtmpMessage。这也是 rtmpServer-master 最容易被新人改坏的地方。常见错误是为了「性能」直接按 TCP 收到的 ByteBuf 切结果遇到一个 chunk 被拆在两个 TCP 包时解析错位或者把所有 chunk stream 共用同一个累积 buffer不同流 ID 的消息互相污染。我自己做的时候会把累积 buffer 放进MapInteger, ByteBufkey 是 chunk stream id每个流独立维护这样音视频消息和控制消息交错到达也不会出现串数据。3. 部署 rtmpServer-master 并推流验证从启动参数到本地 rtmp 测试地址3.1 服务端启动的最小工程端口、线程模型和写缓冲水位不管是开源包还是你自己搭的工程Netty 服务端入口大同小异。一个能跑起来的示例是EventLoopGroup boss new NioEventLoopGroup(1); EventLoopGroup worker new NioEventLoopGroup(0); // 0 表示按 CPU 核数 * 2 创建 try { ServerBootstrap bootstrap new ServerBootstrap(); bootstrap.group(boss, worker) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_REUSEADDR, true) .childOption(ChannelOption.TCP_NODELAY, true) .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(256 * 1024, 1024 * 1024)) .childHandler(new RtmpServerInitializer()); Channel channel bootstrap.bind(1935).sync().channel(); channel.closeFuture().sync(); } finally { boss.shutdownGracefully(); worker.shutdownGracefully(); }boss线程数固定 1因为 accept 连接不耗多少资源worker是处理读写事件的推流场景 I/O 密集默认 CPU 核数乘 2 就够不需要为了追求并发盲目调高反而会造成上下文切换开销。TCP_NODELAY必须开直播流对延迟敏感不开的话小包会被 Nagle 算法攒住播放端延迟肉眼可见。WRITE_BUFFER_WATER_MARK很关键Netty 默认高水位是 64KB对视频流来说太低了但也不要无脑调大这里的高水位决定写缓冲积压多少就触发 channel 不可写我一般设成 256KB 到 1MB具体看你单路平均码率避免一个慢播放端把整个服务端内存拖爆。3.2 用 ffmpeg 对本地 rtmp 测试地址推流验证 publish 链路服务端启动后最直接的动作就是用 ffmpeg 推一路流上去。先用一个真实文件的推流命令理清流程ffmpeg -re -stream_loop -1 -i sample.mp4 -c copy -f flv rtmp://127.0.0.1:1935/live/stream1-re让 ffmpeg 按原帧率读取模拟实时推流-stream_loop -1循环推同一个文件-c copy不转码直接把 sample.mp4 的编码数据塞进 FLV 容器-f flv指定输出封装RTMP 推流本质上就是 FLV over RTMP。后面这个地址就是本地 rtmp 测试地址127.0.0.1:1935是服务端监听地址live是 app 名stream1是流的标识。在 RTMP 命令里connect的 app 参数是livepublish的名字是stream1这两个值要能对得上。如果没有现成视频文件可以用 ffmpeg 合成测试源省去找素材的麻烦ffmpeg -re -f lavfi -i testsrc2size640x360:rate30 \ -f lavfi -i sinefrequency440 \ -vcodec libx264 -acodec aac -f flv rtmp://127.0.0.1:1935/live/test推流开始后服务端日志里应该依次出现connect、createStream、publish这类命令记录。如果只在 codec 里看到字节被消费却没有任何命令输出问题大概率出在握手或 chunk 解码而不是业务 handler。3.3 推流没反应时先抓包确认握手和 AMF 命令真的发出去了RTMP 服务端有个特点出了问题像黑匣子日志不打印你根本不知道卡在哪一步。我遇到这种情况会先用 tcpdump 抓本地回环包tcpdump -i lo -s 0 -X -nn port 1935重点看两处。第一是握手字节第一条 TCP 数据应该以0x03开头后面跟 1536 字节的 C1如果0x03之后没有后续数据往往是服务端没有及时返回 S0S1S2。第二是 connect 命令AMF0 字符串的开头固定是0x02 0x00 0x0c后面跟着connect的 ASCII 字节。如果你能在抓包里看到这些说明 chunk 层已经通了如果只能看到乱码查一下 chunk decoder 是不是把控制消息的 chunk size 也误当成媒体消息处理了。4. 抄作业Netty 里实现 RTMP publish 的核心代码与参数调整4.1 握手响应怎么回版本检查、S1 随机串和 S2 回显 C1握手是最容易写假的地方。很多简化实现直接忽略 C1 内容随意回一个 1536 字节的缓冲区客户端比如 ffmpeg 通常也能蒙混过关但遇到严格校验的客户端就会断连。标准做法是S1 由时间戳、零和随机字节组成S2 回显对端 C1 的全部 1536 字节。下面的代码是一个可以直接抄进 Netty ByteToMessageDecoder 的最小握手实现public class RtmpHandshakeDecoder extends ByteToMessageDecoder { private static final int HANDSHAKE_LEN 1536; private boolean handshakeReplied false; Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, ListObject out) { // 第一次等待 C0 C1 if (!handshakeReplied) { if (in.readableBytes() 1 HANDSHAKE_LEN) { return; // 数据不够继续等待半包 } byte version in.readUnsignedByte(); if (version ! 0x03) { ctx.close(); // RTMP 目前只有 0x03 这个版本 return; } byte[] c1 new byte[HANDSHAKE_LEN]; in.readBytes(c1); ByteBuf s1 ctx.alloc().buffer(HANDSHAKE_LEN); s1.writeInt((int) (System.currentTimeMillis() / 1000)); // S1 时间戳 s1.writeInt(0); // 全零 s1.writeBytes(createRandom(HANDSHAKE_LEN - 8)); // S2 按规范回显 C1 的全部字节 ByteBuf s2 ctx.alloc().buffer(HANDSHAKE_LEN); s2.writeBytes(c1); ByteBuf response ctx.alloc().buffer(1 HANDSHAKE_LEN * 2); response.writeByte(0x03); response.writeBytes(s1); response.writeBytes(s2); s1.release(); s2.release(); ctx.writeAndFlush(response); handshakeReplied true; return; } // 第二次等待 C2C2 内容通常不必校验 if (in.readableBytes() HANDSHAKE_LEN) { return; } in.skipBytes(HANDSHAKE_LEN); // 如果 C2 和后续 connect 命令在同一批数据里剩余部分继续往下传 if (in.readableBytes() 0) { out.add(in.readRetainedSlice(in.readableBytes())); } } }createRandom是生成随机字节的辅助函数可以用SecureRandom也可以用ThreadLocalRandom拼出来。这里有两个参数容易踩坑HANDSHAKE_LEN 必须是 1536这是 RTMP 固定值S0 的0x03不能丢否则客户端会认为版本不支持。握手完成后把out.add剩下的字节交给下个 handler这正好对应第 2 章里 pipeline 的顺序。4.2 AMF0 命令解析connect、createStream、publish 怎么回才让 ffmpeg 继续推握手之后客户端会先发connect再发createStream最后发publish。服务端状态机要按顺序响应。RTMP 命令是 message type 20 的 payload里面用 AMF0 编码。解析命令名和事务 ID 的代码可以封装成一个小工具public static RtmpCommand readCommand(ByteBuf payload) { // 第一个成员是命令名字符串 byte type payload.readUnsignedByte(); if (type ! 0x02) { throw new RtmpException(命令名必须是 AMF0 string实际 type type); } int nameLen payload.readUnsignedShort(); if (nameLen payload.readableBytes()) { throw new RtmpException(AMF0 字符串长度越界: nameLen); } String command payload.readCharSequence(nameLen, StandardCharsets.UTF_8).toString(); // 第二个成员是事务 IDAMF0 number 是 8 字节 double payload.readUnsignedByte(); // 跳过 0x00 number marker long transactionId (long) payload.readDouble(); return new RtmpCommand(command, transactionId, payload); }真正影响推流能不能继续的是回复格式。connect的回复必须带相同的 transactionId返回_resultcreateStream的回复除了_result还要在后面附一个 stream id否则 ffmpeg 拿不到有效的流 ID会一直停在等待状态。publish 命令本身可以不回 onFCPublish很多实现只回一个_resultffmpeg 也不再纠结但按协议规范最好回onFCPublish再回_result播放端将来做状态展示时更正确。回复 payload 的 AMF0 编码最容易漏写的是每个成员前面的 type marker。字符串前要写0x02数字前要写0x00。我见过有人把_result前面的0x02漏掉导致 ffmpeg 直接把整个命令当无效数据丢弃表现为推流端一直卡在连接阶段服务端却日志正常。遇到这种问题最有效的排查办法是在发送前把 payload 的 HEX dump 打出来对比字符串开头是否真的是0200005f 72657375 6c74这种模式。4.3 chunk size 协商与媒体数据转发既要低开销也要防别人不消费RTMP 默认 chunk size 是 128 字节这对服务端是很不友好的。客户端推流时通常会主动发一个 Set Chunk Size 控制消息把 chunk size 调到 4096 或更大。这个 control message 的 message type ID 是 5服务端收到后要记住当前连接的新 chunk size之后解析继续按新值拆块。同样服务端给播放端推流时也可以主动发 Set Chunk Size减少拆包数量降低 CPU 开销。收到 publish 之后的 audio/video 消息要做的是把 payload 原样分发给房间里的订阅者。这里最容易犯的错是给每个订阅者都重新拷贝一份完整 ByteBuf。正确做法是用retain()增加引用计数让多个输出共享同一个 payload写完后由 Netty 的写回调统一释放private final MapInteger, ChannelGroup rooms new ConcurrentHashMap(); Override protected void channelRead0(ChannelHandlerContext ctx, RtmpMessage msg) { if (msg.getTypeId() 8 || msg.getTypeId() 9) { ChannelGroup subscribers rooms.get(msg.getStreamId()); if (subscribers null) { // 没人播放时直接丢弃不要把视频帧累积在内存里 return; } ByteBuf payload msg.getPayload().retain(); RtmpMessage copy RtmpMessage.media(msg.getTypeId(), msg.getTimestamp(), msg.getStreamId(), payload); subscribers.writeAndFlush(copy); } }ChannelGroup来自 Netty 的io.netty.channel.group它自动管理连接关闭时移除订阅者省去手动遍历。这里有一个参数值得单独说如果直播间没有任何播放者队列会一直为空此时媒体帧必须直接丢如果你为了「秒开」保留最近一个关键帧注意要定时清理否则一堆无人观看的流会把内存吃掉。5. Netty 做 RTMP 服务端最容易踩的 5 个坑从粘包处理到连接泄漏排查5.1 粘包半包为什么 ByteToMessageDecoder 还会解析错位现象推流一段时间后命令解析出现乱码message type ID 对不上甚至出现奇怪的异常。原因RTMP 不存在按消息对齐的 TCP 包一个 chunk 可能跨多个 TCP 包多个 chunk 也可能在同一次 read 中到达。如果只在 chunk 解码器里简单地循环readByte()没有先检查 basic header 的长度和当前 payload 剩余字节半包时就会把半个 chunk 当成完整 chunk 处理。Netty 的ByteToMessageDecoder只是帮你把 TCP 半包攒到一起并不能替代协议层的边界判断。解决在解码循环里先读 basic header 的第一个字节从低 6 位拿到 chunk stream id如果 csid 是 0 或 1basic header 会延长到 2 或 3 字节。然后根据 fmt 判断 message header 长度按 message length 累积。常见做法是每个 chunk stream id 维护一个ByteBuf只有当可读字节数 当前 chunk 剩余长度时才消费否则直接 return 等下一次 channelRead。不要用getByte连续探测而不移动 readerIndex那样在数据不足时会重复处理同一个字节。5.2 大关键帧被拆成多个 chunk服务端报「消息不完整」现象直播画面卡住或断流日志中出现incomplete message或者播放端花屏后连接被服务端关闭。原因视频关键帧动辄几十 KB默认 128 字节的 chunk size 会把它切成一长串块。如果解码器没有按 message length 做累计或中途插入了控制消息导致累积 buffer 被重置就会永远等不到完整 payload。另一个隐蔽场景是同一个 chunk stream 里既发了控制消息又发了媒体消息状态没分开导致头部读错。解决为每个 chunk stream 维护partialMessage对象包含messageLength、expectedChunkLength和累积用的ByteBuf。收到新 chunk 时先比较剩余可读字节如果不够留下当前 chunk 已读到的部分并返回够才继续读。建议给messageLength设一个上限比如 10MB超过直接断开这条推流流避免恶意构造超大消息把内存撑挂。5.3 推流端异常断开后连接数只涨不降现象用lsof -i :1935看连接数持续增长几天后服务端报Too many open files。原因推流端拔网线、杀进程或断网TCP 不会立刻发 FIN服务端一直以为连接还活着。如果 handler 里没有处理channelInactivepipeline 里也没有空闲检测这些半开连接会一直占用 fd 和 buffer。RTMP 推流场景尤其常见因为 OBS 或 ffmpeg 异常退出时不一定有完整挥手。解决我在RtmpServerInitializer里加了IdleStateHandler(60, 0, 0, TimeUnit.SECONDS)并在业务 handler 里覆盖userEventTriggered收到IdleStateEvent时直接ctx.close()。同时在channelInactive中把该连接对应的发布流、房间订阅关系和残留 ByteBuf 全部清理不要让ChannelGroup里留着已关闭的 channel。5.4 AMF0 解析越界字符串长度与可读字节不一致现象高并发推流时偶发IndexOutOfBoundsException堆栈指向readUnsignedShort或readBytes。原因在 chunk 还没拼完整 payload 时就开始解析 AMF0。AMF0 字符串前两字节是长度如果这两字节已经到达后面的字符串内容还没到代码按长度去读就会越过可读边界。另一个原因是把 AMF0 解析放在 chunk 解码器内部不同 chunk 之间的数据被错误拼接。解决把 AMF0 解析放到RtmpMessageDecoder这一层此时RtmpChunkDecoder已经保证输出的是一个完整消息payload 不再有半包问题。在解析函数开头仍要判断payload.readableBytes() 3时提前返回或抛出业务异常不要假设数据一定够。字符串长度读取后立刻和readableBytes()比较越界就记录日志并关闭连接。5.5 时间戳回绕和长时间直播断流现象直播跑了一天多后画面突然卡住ffmpeg 侧提示 timestamp 异常播放端时间线乱跳。原因RTMP 时间戳是 24 位无符号整数最大 0xFFFFFF超过之后会回绕到 0。服务端如果用有符号int直接转发回绕时会出现负数播放器无法处理。解决为每个流维护一个基准时间戳收到新时间戳后计算差值差值小于 0 时加上0x1000000再继续累加。我在转发音频和视频时都会走这个逻辑并把这个差值缓冲到该流的状态里。排查这种问题不要只看当前时间戳要把前后两条消息的差值打出来差值连续为负基本就是 24 位回绕问题。6. 从「能推流」到「抗并发」验证手段、压测参数与 WHIP 对照新代码上线前先不要急着加功能。用 10 路 ffmpeg 并发推流跑 1 小时再观察服务端内存曲线这是我最常用的第一步验证。命令可以写成一个循环每组推不同的 stream keyfor i in $(seq 1 10); do ffmpeg -re -f lavfi -i testsrc2size320x180:rate15 \ -f lavfi -i sinefrequency440 \ -vcodec libx264 -preset ultrafast -acodec aac \ -f flv rtmp://127.0.0.1:1935/live/stream$i done跑起来后用jstat -gcutil pid 1000盯着 Full GC。如果老年代持续增长优先怀疑有 handler 里retain()后没有 release或者房间 map 里残留了已关闭 channel。Netty 的堆外内存不显示在堆里还要看/proc/pid/status里的 VmRSS增长过快多半是 DirectByteBuffer 泄漏重点是检查写出去的ByteBuf在回调里有没有释放。压测到 100 路以上时worker线程数保持默认即可最需要调的是WRITE_BUFFER_WATER_MARK。把低水位设成单路平均码率的 1 到 2 倍高水位设成低水位的 3 到 4 倍这样 Netty 在某个订阅者消费不及时的时候会触发可写状态反转而不是无限积压。另一个要防的是connect后长时间不发publish的连接这类连接会占着流 ID 和线程资源加一个「连接建立后 10 秒内必须完成 publish否则断开」的超时很实用。做久了你会发现RTMP 的对手不是 SRT而是 WHIP 推流。浏览器直接推流到服务器的场景里WHIP/WebRTC 延迟能压到 500ms 以内RTMP 再优化也基本在 1 秒以上。但 WHIP 的工程复杂度高得多涉及 ICE、DTLS、SRTP重协商也不如 RTMP 简单。我现在的习惯是对外兼容老设备、OBS 和已有推流端时就保留 rtmpServer 这一套先让它足够稳定等确实需要浏览器低延迟推流时再在旁边挂一个 WHIP 网关把两边用消息队列串起来而不是在同一个 Netty pipeline 里强行混协议。配置参数、超时时间这些都要在测试环境压过一遍再上生产不然现场调玄学参数很难收场。希望这篇能帮你少踩几个坑把 rtmpServer-master 真正用起来。本文还有配套的精品资源点击获取
