推送远端视频流
准备
鉴于ffserver在FFMpeg 3.4版本后就已经被移除（参见这里），FFMpeg无法在没有一个服务器程序协同的情况下单独完成推流工作。同样的问题也存在于mpegCoder中。用户需要先启动一个服务器程序，该程序会持续侦听、等待推送的视频流。在此之后mpegCoder就可以通过mpegCoder.MpegServer推送视频了。
caution
实际上，你也可以在使用MpegServer推送视频的同时，用mpegCoder.MpegClient接收这个视频流。但是我们还是建议用户尽可能在两台不同的机器上运行MpegServer和MpegClient。因为MpegServer自带的编码器会占用很多系统资源。
建议使用以下视频服务器项目。用户按自己的需求，从中选择一个。
| 项目 | Windows | Linux |
|---|---|---|
| 简单RTSP服务器（RTSP Simple Server） | ||
| 马特罗斯卡服务Mk2（Matroska Server Mk2） | ||
| 简单实时服务器（Simple Realtime Server） |
以Windows平台和简单RTSP服务器（RTSP Simple Server）为例，我们只需要通过一行命令启动这个服务器程序即可：
当服务器处于侦听状态时，我们可以使用以下地址来进行推流测试。
rtsp://localhost:8554/
rtmp://localhost:1935/
范例：非阻塞式推流
此例基于非阻塞式API MpegServer.ServeFrame()。在推流的过程中，确保数据同步是一个很重要的问题。如果我们一直不断地使用ServeFrame()，那么我们就会尽可能地能推送多少帧、就推送多少帧。这些新推送的帧就会覆盖掉之前推送的帧。在一些情况下，服务器甚至会崩溃，因为服务器无法接收如此多的帧。
为了保证服务器能正常运转，我们需要按照视频的时间戳来推送帧。当MpegServer.FFmpegSetup()成功调用时，将会设置一个开始时间戳。MpegServer会维护一个计时器，每当用户调用MpegServer.getParemeter('waitRef')时，该方法就会返回一个推荐等待时长，用来表示推送出去的视频帧已经比实际视频帧多出了多久。这个推荐等待时长就是上述的这个时间间隔的一半（单位为秒）。如果我们推送了过多帧，就可以利用这个参数让服务等待一会。
import time
import mpegCoder
d = mpegCoder.MpegDecoder()
opened = d.FFmpegSetup('test-video.mp4')
e = mpegCoder.MpegServer()
e.setParameter(configDict=d.getParameter(), codecName='libx264', videoAddress='rtsp://localhost:8554/video') # 从解码器继承绝大多数设置。
opened = opened and e.FFmpegSetup() # 加载推流器。
if opened: # 如果推流器、解码器没有正常载入，停止后续步骤。
gop = True
s = 0
while gop is not None:
gop = d.ExtractGOP() # 提取当前的画面组。
if gop is not None:
for i in gop: # 遍历每一帧。
e.ServeFrame(i) # 编码并推送当前帧。
s += 1
if s == 10: # 每过10帧，检查、并等待播放同步。
wait = e.getParameter('waitRef')
time.sleep(wait)
s = 0
e.FFmpegClose() # 结束编码和推流，并将缓存内的所有帧刷入视频流内。
else:
print(e)
e.clear() # 清除推流器设置。
d.clear() # 关闭解码器、并清除设置。
范例：双进程模式
以上的例子并不是一个优雅的实现，因为MpegDecoder和MpegServer同时抢占了主线程。如果解码器需要花费相当的时间，那么推流就会出现明显延迟。因此，建议将MpegDecoder和MpegServer分离到两个不同的子进程里。下面的代码就是通过这种方式实现的。解码器和推流器通过一个共享的数据队列实现同步。在此我们使用MpegServer.ServeFrameBlock()取代MpegServer.ServeFrame()。每当调用这个方法的时候，MpegServer就会检查当前的播放时长，并自动确保新推送帧的时间戳不超过播放时长过多。如果新帧的时间戳和播放时长的差距过大，该方法就会阻塞所在的线程，直到这个差距小到可以接受为止。
import mpegCoder
import multiprocessing
class Decoder(multiprocessing.Process):
def __init__(self, video_name='test-video.mp4', q_o=None, name=None, daemon=None):
super().__init__(name=name, daemon=daemon)
self.video_name = video_name
self.q_o = q_o
def run(self):
d = mpegCoder.MpegDecoder()
opened = d.FFmpegSetup(self.video_name)
self.q_o.put(d.getParameter())
if opened:
gop = True
while gop is not None:
gop = d.ExtractGOP() # 提取当前的画面组。
if gop is not None:
for i in gop: # 遍历每一帧。
self.q_o.put(i)
else:
self.q_o.put(None)
else:
print(d)
d.clear()
class Encoder(multiprocessing.Process):
def __init__(self, video_addr='rtsp://localhost:8554/video', q_i=None, name=None, daemon=None):
super().__init__(name=name, daemon=daemon)
self.video_addr = video_addr
self.q_i = q_i
def run(self):
e = mpegCoder.MpegServer()
config_dict = self.q_i.get() # 获取解码器的参数。
e.setParameter(configDict=config_dict, codecName='libx264', maxBframe=16, videoAddress=self.video_addr)
opened = e.FFmpegSetup()
if opened: # 如果推流器没有正常加载，就停止以下步骤。
frame = True
while frame is not None:
frame = self.q_i.get() # 获取一帧。
if frame is not None:
e.ServeFrameBlock(frame) # 编码并推送当前帧。
e.FFmpegClose() # 结束编码和推流，并将缓存内的所有帧刷入视频流内。
else:
print(e)
e.clear()
if __name__ == '__main__':
queue_data = multiprocessing.Queue(maxsize=20)
proc_dec = Decoder(video_name='test-video.mp4', q_o=queue_data, daemon=True)
proc_enc = Encoder(video_addr='rtsp://localhost:8554/video', q_i=queue_data, daemon=True)
proc_dec.start()
proc_enc.start()
proc_enc.join()
proc_dec.join()
caution
在上例里，调用MpegServer.setParameter()时使用了configDict。这个输入值是MpegDecoder.getParameter()返回的一个python字典。该用法等价与使用e.setParameter(decoder=d)。然而，此例中我们必须使用这个等价用法，因为mpegCoder所有的实例都无法被pickled。