之前几篇文章我们了解了多媒体数据流的采集和编码，这一篇开始我们来了解数据传输的流程。
在数据流编码完成以后，程序会通过打包器将编码完成的数据打包再传输出去，代码如下：
// The packetizer encapsulates the bit stream in an RTP stream and send it over the network
mPacketizer.setDestination(mDestination, mRtpPort, mRtcpPort);
mPacketizer.setInputStream(new MediaCodecInputStream(mMediaCodec));
mPacketizer.start();
简单来说，就是设置传输目的地地址、Rtp端口和Rtcp端口，设置数据源的InputStream(MediaCodecInputStream继承InputStream，从MediaCodec对象中获取完成编码的数据流)，启动start()方法执行数据打包操作。下面我们来看一下打包器的源码：
AbstractPacketizer类
AbstractPacketizer就是打包器的基类，封装了对数据流打包操作的基本操作和一些公共参数变量。
public AbstractPacketizer() {
int ssrc = new Random().nextInt();
ts = new Random().nextInt();
socket = new RtpSocket();
socket.setSSRC(ssrc);
}
...
/** Starts the packetizer. */
public abstract void start();
/** Stops the packetizer. */
public abstract void stop();
/** Updates data for RTCP SR and sends the packet. */
protected void send(int length) throws IOException {
socket.commitBuffer(length);
}
上面截取AbstractPacketizer类的部分代码。AbstractPacketizer的构造函数中直接创建了一个RtpSocket对象，就是将打包好的数据用Rtp协议传输出去的执行者。ssrc：用于标识同步信源。ts：时间戳。AbstractPacketizer还提供了两个抽象方法start()和stop()，用于控制流的打包操作。在子类（根据数据格式生成不同的子类）实现的start()方法中，会创建一个线程来执行数据打包操作，数据打包的内部原理涉及到音视频的相关格式和相关传输协议，这里不再深入。send(int length)方法就是使用RtpSocket对象更新和发送数据包。
RTP协议和RTCP协议
- RTP全名是Real-time Transport Protocol（实时传输协议），RTCP（Real-time Transport Control Protocol，即实时传输控制协议）。
- RTP用来为IP网上的语音、图像、传真等多种需要实时传输的多媒体数据提供端到端的实时传输服务。
- RTP为Internet上端到端的实时传输提供时间信息和流同步，但并不保证服务质量，服务质量由RTCP来提供。
- RTCP的主要功能是：服务质量的监视与反馈、媒体间的同步，以及多播组中成员的标识。在RTP会话期 间，各参与者周期性地传送RTCP包。RTCP包中含有已发送的数据包的数量、丢失的数据包的数量等统计资料，因此，各参与者可以利用这些信息动态地改变传输速率，甚至改变有效载荷类型。RTP和RTCP配合使用，它们能以有效的反馈和最小的开销使传输效率最佳化，因而特别适合传送网上的实时数据。
- Rtp和Rtcp分别使用两个端口执行通信。RTP数据发向偶数的UDP端口，而对应的控制信号RTCP数据发向相邻的奇数UDP端口（偶数的UDP端口＋1），这样就构成一个UDP端口对。
- 参考资料：RTP协议分析 应该稍微了解一下RTP的封装和RTCP的封装。
RtpSocket类
RtpSocket类是使用RTP协议的Socket的封装实现。
/**
* This RTP socket implements a buffering mechanism relying on a FIFO of buffers and a Thread.
* @throws IOException
*/
public RtpSocket() {
mCacheSize = 00;
mBufferCount = 300; // TODO: reajust that when the FIFO is full
mBuffers = new byte[mBufferCount][];
mPackets = new DatagramPacket[mBufferCount];
mReport = new SenderReport();
mAverageBitrate = new AverageBitrate();
...
try {
mSocket = new MulticastSocket();
} catch (Exception e) {
throw new RuntimeException(e.getMessage());
}
}
在构造函数中初始化各个参数变量和发送RTCP报文的实例（SenderReport），mSocket对象（这里使用MulticastSocket，实现将数据报以广播的方式发送到多个client）。
/** Sends the RTP packet over the network. */
public void commitBuffer(int length) throws IOException {
updateSequence();
mPackets[mBufferIn].setLength(length);
mAverageBitrate.push(length);
if (++mBufferIn>=mBufferCount) mBufferIn = 0;
mBufferCommitted.release();
if (mThread == null) {
mThread = new Thread(this);
mThread.start();
}
}
发送RTP协议包数据。这里启动了一个线程来执行发送动作，以一定的速率依次发送数据包。
/** The Thread sends the packets in the FIFO one by one at a constant rate. */
@Override
public void run() {
Statistics stats = new Statistics(50,3000);
try {
// Caches mCacheSize milliseconds of the stream in the FIFO.
Thread.sleep(mCacheSize);
long delta = 0;
while (mBufferCommitted.tryAcquire(4,TimeUnit.SECONDS)) {
if (mOldTimestamp != 0) {
// We use our knowledge of the clock rate of the stream and the difference between two timestamps to
// compute the time lapse that the packet represents.
if ((mTimestamps[mBufferOut]-mOldTimestamp)>0) {
stats.push(mTimestamps[mBufferOut]-mOldTimestamp);
long d = stats.average()/1000000;
//Log.d(TAG,"delay: "+d+" d: "+(mTimestamps[mBufferOut]-mOldTimestamp)/1000000);
// We ensure that packets are sent at a constant and suitable rate no matter how the RtpSocket is used.
if (mCacheSize>0) Thread.sleep(d);
} else if ((mTimestamps[mBufferOut]-mOldTimestamp)<0) {
Log.e(TAG, "TS: "+mTimestamps[mBufferOut]+" OLD: "+mOldTimestamp);
}
delta += mTimestamps[mBufferOut]-mOldTimestamp;
if (delta>500000000 || delta<0) {
//Log.d(TAG,"permits: "+mBufferCommitted.availablePermits());
delta = 0;
}
}
mReport.update(mPackets[mBufferOut].getLength(), System.nanoTime(),(mTimestamps[mBufferOut]/100L)*(mClock/1000L)/10000L);
mOldTimestamp = mTimestamps[mBufferOut];
if (mCount++>30) mSocket.send(mPackets[mBufferOut]);
if (++mBufferOut>=mBufferCount) mBufferOut = 0;
mBufferRequested.release();
}
} catch (Exception e) {
e.printStackTrace();
}
mThread = null;
resetFifo();
}
简单流程就是：配置传输速率，使用while来循环执行，如果一个执行过程不超过4秒，则一直循环下去。在循环执行的动作包括：计算速率的平均值（按照这个平均值的速率来执行传输，确保速率恒定），更新Rtcp报文，Socket执行发送动作(Send)来给客户端（接收端）传输数据包。
SenderReport类
SenderReport类是执行Rtcp协议的实现类。
public SenderReport() {
...
try {
usock = new MulticastSocket();
} catch (IOException e) {
throw new RuntimeException(e.getMessage());
}
upack = new DatagramPacket(buffer, 1);
// By default we sent one report every 5 secconde
interval = 3000;
}
正如上面介绍时所说，Rtp和Rtcp分别使用两个端口执行通信，所以SenderReport类的构造函数也需要初始化一个Socket对象用于发送Rtcp报文。
/**
* Updates the number of packets sent, and the total amount of data sent.
* @param length The length of the packet
* @throws IOException
**/
public void update(int length, long ntpts, long rtpts) throws IOException {
packetCount += 1;
octetCount += length;
setLong(packetCount, 20, 24);
setLong(octetCount, 24, 28);
now = SystemClock.elapsedRealtime();
delta += oldnow != 0 ? now-oldnow : 0;
oldnow = now;
if (interval>0) {
if (delta>=interval) {
// We send a Sender Report
send(ntpts,rtpts);
delta = 0;
}
}
}
在Rtp的Socket不断发送数据包的同时，SenderReport也不断在更新和发送Rtcp的报文，
update()方法就是在不断更新发送的数据包数，以及发送的数据总量。
/** Sends the RTCP packet over the network. */
private void send(long ntpts, long rtpts) throws IOException {
long hb = ntpts/1000000000;
long lb = ( ( ntpts - hb*1000000000 ) * 4294967296L )/1000000000;
setLong(hb, 8, 12);
setLong(lb, 12, 16);
setLong(rtpts, 16, 20);
upack.setLength(28);
usock.send(upack);
}
send()方法就是用于在update()更新Rtcp报文后，将新的Rtcp报文通过Socket传给接收端。
到这里我们了解了打包器Packetizer的打包流程和Rtp&Rtcp协议的传输流程，下一篇将了解Rtsp协议和它在项目中的运用流程。
网友评论