波仔
🎬 第八章 FFmpeg命令行实现流媒体功能及直播应用
📄 一句话概括

推流端与拉流端是两条互不相识的 TCP 连接，只在服务器内存里交汇。 观众来来去去，主播一个包都不少发。

📝 本节和前面几节的关系

8.6.1 讲协议长什么样，这一节讲三个角色怎么配合。 8.6.3 与 8.6.4 再分别细看两端。

一、三个角色，两条连接

一场直播里只有三个角色，但连接永远只有两条（每个观众各一条）。

角色 谁发起 发什么 收什么

推流端 客户端主动连 publish ＋ 媒体 几乎没有 服务器 被动 accept 转发 — 拉流端 客户端主动连 play 媒体流

关键点：推流端与拉流端完全不知道对方的存在。 它们唯一的交集，是服务器内存里的那个「流」对象。

所以主播掉线，观众只是画面停住；观众全走光，主播照推不误。 这种「两端解耦」是流媒体服务器存在的全部理由。

二、一条流能同时喂几个观众

服务器干的事说穿了就一句：收一份，复制 N 份。 下面这个实验让推流端发 20 个包，服务器转发给 3 个观众。

观众 收到字节 该收多少

1 号 20000 20000 2 号 20000 20000 3 号 20000 20000

服务器一共转发 60 次，正好是 20 个包乘 3 个观众。 推流端只发了一份，出口这边变成了三份——这就是「一进多出」。

算一下带宽：推流端 1 份上行，服务器 3 份下行。 一百个观众就是 100 份下行，所以流媒体服务器最吃的是出口带宽， CPU 反倒不怎么吃——除非它还要转码。

三、观众退场，主播知道吗

再做一个实验：观众看到第 5 个包就关掉连接，看服务器怎么反应。

项 值

推流端发出 10 个 转发成功 6 次 转发失败 4 次

转发成功 6 次、失败 4 次——前几个包还算「写进去了」， 后面才发现对面早就断了（TCP 的 RST 需要一来一回才传到）。

但推流端一个包都没少发，它压根不知道有人在看、更不知道有人走了。 这正是流媒体服务器要替主播挡住的事：观众的生灭不该影响主播。

服务器要做的收尾是：把这个观众的连接从列表里摘掉， 别让写失败拖慢其他观众——这一步叫「踢掉慢消费者」。

四、服务器不在时，推流会怎样

这一条比想象中更值得警惕。

项 值

结果 没报错 耗时 6 ms

打开不报错、mux 不报错、close 也不报错——三处都不吭声。 因为 av.open 是「懒连接」，真正去连是后面的事； 而连不上时那些字节会被直接丢掉，库也不告诉你。

这就是推流最坑的地方：一切看着都成功，服务器那边一根毛都没有。

五、怎么确认「真的推上去了」

既然客户端靠不住，就只能从外面看。

办法 看什么 可靠性

看服务器日志 有没有收流记录 最直接 抓本地端口 1935 有没有流量 很硬 看服务器状态页 SRS 这类自带页面 方便 让播放器试拉 能不能出画面 端到端

四条里最省事的是「让播放器试拉」，因为它把整条链路都验了。 但排查问题时要从最上游往下走： 先确认服务器收到，再看播放器能不能拉到。

六、动手 Demo：起个转发服务器，量一进多出

下面这段用 20 行代码顶替流媒体服务器：一个推流端、三个观众。 第二个实验让观众中途退场，看推流端有没有受影响。

import socket, threading, time

def free_tcp():
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    p = s.getsockname()[1]
    s.close()
    return p

NCH, SIZE = 20, 1000
port = free_tcp()
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(("127.0.0.1", port))
srv.listen(8)
srv.settimeout(8.0)
stat = {"got": [], "fwd": 0, "dropped": 0}

def serve():
    pusher, _a = srv.accept()
    viewers = [srv.accept()[0] for _i in range(3)]
    for _ in range(NCH):
        d = pusher.recv(SIZE + 16)
        if not d:
            break
        for v in viewers:
            try:
                v.sendall(d)          # 一份变三份
                stat["fwd"] += 1
            except Exception:
                stat["dropped"] += 1
    for v in viewers:
        v.close()
    pusher.close()

threading.Thread(target=serve, daemon=True).start()
time.sleep(0.2)
pusher = socket.create_connection(("127.0.0.1", port))
viewers = [socket.create_connection(("127.0.0.1", port)) for _i in range(3)]

def watch(v):
    n = 0
    v.settimeout(6.0)
    try:
        while True:
            d = v.recv(4096)
            if not d:
                break
            n += len(d)
    except Exception:
        pass
    stat["got"].append(n)

for v in viewers:
    threading.Thread(target=watch, args=(v,), daemon=True).start()
for _i in range(NCH):
    pusher.sendall(b"z" * SIZE)
    time.sleep(0.005)
time.sleep(1.0)
print(u"每位观众收到 %s" % stat["got"])
print(u"服务器转发 %d 次，失败 %d 次" % (stat["fwd"], stat["dropped"]))

实跑输出：

--------------------------------------------------------------------
8.6.2 直播推流与拉流
--------------------------------------------------------------------

**① 三个角色，两条连接**

  角色       谁发起           发什么              收什么
  ------------------------------------------------------------
  推流端     客户端主动连     publish ＋ 媒体     几乎没有
  服务器     被动 accept      转发                —
  拉流端     客户端主动连     play                媒体流

**注意：推流端与拉流端是两条独立的 TCP 连接，互不相识。**
**它们唯一的交集是服务器内存里的那个「流」。**

**② 实测：一条流能同时喂给三个观众**

  观众     收到字节     该收多少
  ----------------------------------
  1 号     20000        20000
  2 号     20000        20000
  3 号     20000        20000

**服务器一共转发 60 次，等于 20 个包乘 3 个观众；失败 0 次。**
**推流端只发了一份，服务器复制了三份——这就是流媒体服务器的全部价值。**
**观众越多，服务器出口带宽越大，CPU 反倒不怎么吃。**

**③ 实测：观众中途退场，推流端受影响吗**

  项             值
  ------------------------
  推流端发出     10 个
  转发成功       6 次
  转发失败       4 次

**推流端一个包都没少发——它压根不知道有人在看、更不知道有人走了。**
**这是流媒体服务器的关键设计：观众的生灭不影响主播。**

**④ 实测：推流时服务器不在，客户端要等多久**

  项       值
  -------------------
  结果     没报错
  耗时     6 ms

**打开不报错、mux 不报错、close 也不报错——三处都不吭声。**
**因为 av.open 是「懒连接」，真正去连是后面的事；**
**而连不上时那些字节会被直接丢掉，库也不告诉你。**

**这就是推流最坑的地方：一切看着都成功，服务器那边一根毛都没有。**
**所以「打开成功」在推流里一点意义都没有，必须另有手段确认。**

三位观众各收 20000 字节，转发 60 次、失败 0 次——一进多出成立。

注意 stat["fwd"] 数的是「写次」，不是「字节」： 20 个包乘 3 个观众等于 60 次写，这个数字最容易看错。

30 秒实验

  1. 把观众从 3 个加到 6 个：看转发次数是不是跟着翻倍。
  2. 让其中一个观众只读不关：看服务器会不会被它拖慢。
  3. 把 sendall 换成 send：可能只写一半，体会一下「部分写」。

常见误区

  1. 以为推流与拉流是同一个连接。它们是两条，靠服务器内存里的流关联。
  2. 以为观众掉线会通知主播。推流端完全不知道，这是设计如此。
  3. 以为 av.open 成功就是推上去了。连不上也照样返回成功。
自测三问（点开对答案） 1. 一场直播里，推流端与拉流端是怎么关联起来的？ —— 没有任何直接关系，只通过服务器内存里的那个「流」对象间接关联。 2. 20 个包发给 3 个观众，服务器要写多少次？ —— 60 次。每收一个包就要给每个观众各写一次。 3. 为什么「打开成功」不能证明推流成功？ —— av.open 是懒连接，连不上也返回成功，字节被静默丢弃。

相关笔记