传输H.264的组播RTSP服务
参考：从零开始写一个RTSP服务器（八）一个多播的RTSP服务器
h264文件：test.h264文件地址
和java实现传输H.264的RTSP服务区别
- 服务端往组播ip+port发送H.264Rtp数据，循环发送
- rtsp的响应：DESCRIBE和SETUP修改
代码
-
RtspTcpServer.java
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Arrays;
import java.util.Date;
// Linux内核对TCP连接的识别是通过四元组来区分：源ip，源port，目标ip，目标port
public class RtspTcpServer {
public static void main(String[] args) throws IOException, InterruptedException {
int rtcpPort = MulticastServer.start();//开启往组播组发送数据
ServerSocket serverSocket = new ServerSocket(8888);//1.创建服务端对象
System.out.println("TCP服务端端启动===>"+serverSocket.getLocalSocketAddress());
while (true){
Socket socket = serverSocket.accept(); //阻塞式,2.获取连接过来的客户端对象
//获取到连接，则开启一个线程处理当前连接
new Thread(new Runnable() {
@Override
public void run() {
InputStream inputStream = null;
OutputStream outputStream = null;
try {
System.out.println("TCP已连接===>"+socket.getRemoteSocketAddress());
inputStream = socket.getInputStream();//3.通过socket对象获取输入流,要读取客户端发来的数据
outputStream = socket.getOutputStream();//3.通过socket对象获取输入流,要读取客户端发来的数据
byte[] buffer = new byte[1024*1024];
int readNum = 0;
while((readNum=inputStream.read(buffer))!=-1){
if(readNum>0){
byte[] receive = Arrays.copyOfRange(buffer,0,readNum);
System.out.println("读取的字节数："+readNum);
System.out.println("读取的字节数："+receive.length);
System.out.println("缓冲区大小："+buffer.length);
handlerReceiveData(outputStream,receive,rtcpPort);
}
}
} catch (IOException e) {
e.printStackTrace();
}finally {
System.out.println("断开连接");
if(inputStream!=null){
try {
inputStream.close();
} catch (IOException e) {
e.printStackTrace();
}
}
if(outputStream!=null){
try {
outputStream.close();
} catch (IOException e) {
e.printStackTrace();
}
}
if(socket!=null){
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}).start();
}
}
public static void handlerReceiveData(OutputStream outputStream, byte[] buffer,int rtcpPort){
String receiveStr=new String(buffer);
System.out.println("TCP-----------------接收receiveStr----------------------");
System.out.println(receiveStr);
System.out.println("TCP-----------------接收receiveStr----------------------");
String lines[] = receiveStr.split("\\r?\\n");//按行分割
int cseq=0;
int clientRtpPort=0;
int clientRtcpPort=0;
String url=null;
String localIp=null;
{
for(String line:lines){
if(line.indexOf("rtsp://")>-1){
url = line.split("\\s+")[1];
String[] split = line.split(":");
localIp = split[1].substring(2);
}
if(line.startsWith("CSeq:")){
String[] split = line.split(": ");
cseq = Integer.parseInt(split[1].trim());
}
if(line.startsWith("Transport:")){
String[] split = line.split(";");
for(String i : split){
if(i.startsWith("client_port=")){
String substring = i.substring(12);
String[] split1 = substring.split("-");
clientRtpPort = Integer.parseInt(split1[0].trim());
clientRtcpPort = Integer.parseInt(split1[1].trim());
}
}
}
}
}//获取cseq
String responseStr=null;
if (receiveStr.startsWith("OPTIONS")){
//OPTIONS 请求服务端支持的RTSP方法列表；也可以定时发送这个请求来保活RTSP会话。
responseStr=String.format("RTSP/1.0 200 OK\r\n"+
"CSeq: %d\r\n"+
"Public: OPTIONS, DESCRIBE, SETUP, PLAY\r\n"+
"\r\n",cseq);
}else if(receiveStr.startsWith("SETUP")){
//SETUP：用于配置数据交互的方法（比如制定音视频的传输方式TCP或UDP）。
responseStr=String.format("RTSP/1.0 200 OK\r\n"+
"CSeq: %d\r\n"+
"Transport: RTP/AVP;multicast;destination=%s;source=%s;port=%d-%d;ttl=255\r\n"+
"Session: 66334873\r\n"+
"\r\n",
cseq,
MulticastServer.MULTICAST_ADDRESS,
url,
MulticastServer.PORT,
rtcpPort);
}else if(receiveStr.startsWith("DESCRIBE")){
//DESCRIBE：请求指定的媒体流的SDP描述信息（详细包括音视频流的帧率、编码类型等媒体信息）。
String sdp=String.format("v=0\r\n"+
"o=- 9%d 1 IN IP4 %s\r\n"+
"t=0 0\r\n"+
"a=control:*\r\n"+
"a=type:broadcast\r\n"+
"a=rtcp-unicast: reflection\r\n"+
"m=video %d RTP/AVP 96\r\n"+
"c=IN IP4 %s/255\r\n"+
"a=rtpmap:96 H264/90000\r\n"+
"a=framerate:25\r\n"+
"a=control:track0\r\n",
new Date().getTime(),
localIp,
MulticastServer.PORT,
MulticastServer.MULTICAST_ADDRESS);
//sdp指定多播地址和端口
responseStr=String.format("RTSP/1.0 200 OK\r\nCSeq: %d\r\n"+
"Content-Base: %s\r\n"+
"Content-type: application/sdp\r\n"+
"Content-length: %d\r\n\r\n"+
"%s",
cseq,
url,
sdp.length(),
sdp);
}else if(receiveStr.startsWith("PLAY")){
//PLAY：用于启动(当暂停时重启)交付数据给客户端。
responseStr=String.format("RTSP/1.0 200 OK\r\n"+
"CSeq: %d\r\n"+
"Range: npt=0.000-\r\n"+
"Session: 66334873; timeout=60\r\n" +
"\r\n",
cseq);
}else if(receiveStr.startsWith("PAUSE")){
//PAUSE：用于临时停止服务端的数据的交互（使用PLAY来重新启动数据交互）。
responseStr=String.format("RTSP/1.0 200 OK\r\n" +
"CSeq: %d\r\n" +
"\r\n",cseq);
}else if(receiveStr.startsWith("TEARDOWN")){
//TEARDOWN：请求终止来自服务端的数据的传输。
responseStr=String.format("RTSP/1.0 200 OK\r\n" +
"CSeq: %d\r\n" +
"\r\n",cseq);
}
try {
outputStream.write(responseStr.getBytes());
outputStream.flush();
System.out.println("TCP-----------------响应responseStr----------------------");
System.out.println(responseStr);
System.out.println("TCP-----------------响应responseStr----------------------");
} catch (IOException e) {
e.printStackTrace();
}
}
}
-
MulticastServer.java
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.MulticastSocket;
/*
* IP组播：
* 组播IP地址：用于标识一个IP组播组，范围是从224.0.0.0到239.255.255.255。
* 组播组可以是永久的也可以是临时的：
* - 组播组地址中，有一部分由官方分配的，称为永久组播组。
* - 永久组播组保持不变的是它的ip地址，组中的成员构成可以发生变化。
* - 永久组播组中成员的数量都可以是任意的，甚至可以为零。
* - 那些没有保留下来供永久组播组使用的ip组播地址，可以被临时组播组利用。
* - 224.0.0.0～224.0.0.255为预留的组播地址（永久组地址）。
* - 224.0.1.0～238.255.255.255为用户可用的组播地址（临时组地址），全网范围内有效。
* - 239.0.0.0～239.255.255.255为本地管理组播地址，仅在特定的本地范围内有效。
* 组播源：信息的发送者，组播源不一定属于组播组，它向组播组发送数据，自己不一定是接收者。多个组播源可以同时向一个组播组发送报文。
* 组播组：信息接收者，加入同一组播组的接收者成员可以广泛分布在网络中的任何地方，没有地域限制。
* 组播路由器：支持组播信息传输的所有路由器
*
*
* 1.IP协议规定组播地址的范围是224.0.0.0 ~ 239.255.255.255，协议软件底层设定好的
* 2.java提供了
* - socket=new MulticastSocket(组播port);创建组播成员（port端口）
* - socket.joinGroup(InetAddress.getByName(组播ip));加入组播地址
* - 该socket可以接收发往组播地址的包
* 3.往组播地址发送udp包，所有加入该组播ip+port的成员都能收到
* - byte[] data="哈哈哈".getBytes();
* - DatagramPacket packet=new DatagramPacket(data,data.length,InetAddress.getByName(组播ip),组播port);
* - DatagramSocket datagramSocket=new DatagramSocket();//创建DatagramSocket对象
* - datagramSocket.send(packet);//向目标ip+port端发送数据报
* - datagramSocket.close();
* */
public class MulticastServer {
public static String MULTICAST_ADDRESS = "225.0.0.1";//组播ip
public static int PORT = 6666;//组播端口
private static MulticastSocket multicastSocket;
public static int start() throws IOException {
/*往组播组添加一个成员MulticastSocket，用于发送数据（当然也可以是普通的DatagramSocket）*/
multicastSocket = new MulticastSocket(PORT);
multicastSocket.joinGroup(InetAddress.getByName(MULTICAST_ADDRESS));
/*开启一个专门接收数据的线程（有可能收到谁(-_-)往组播组发的消息）*/
new Thread(new Runnable() {
@Override
public void run() {
while (true) {
try {
byte[] buf = new byte[1024*1024];//创建数据包
DatagramPacket datagramPacket = new DatagramPacket(buf,buf.length);
multicastSocket.receive(datagramPacket);
handlerReceiveData(0,datagramPacket);
} catch (IOException e) {
e.printStackTrace();
}
}
}
}).start();
/*开启一个线程专门往组播组发送h264数据rtp包*/
RTPH264Server rtph264Server = new RTPH264Server(multicastSocket, InetAddress.getByName(MULTICAST_ADDRESS), PORT);
new Thread(new Runnable() {
@Override
public void run() {
try {
rtph264Server.startSendRtpPackage();
} catch (Exception e) {
e.printStackTrace();
}
}
}).start();
/*创建一个rtcp单播Socket*/
DatagramSocket rtcpUdpSocket = new DatagramSocket();//建立socket服务
new Thread(new Runnable() {
@Override
public void run() {
byte[] buf = new byte[1024*1024];//创建数据包
DatagramPacket datagramPacket = new DatagramPacket(buf,buf.length);
while (true){
try {
rtcpUdpSocket.receive(datagramPacket); //阻塞式,3.使用接收方法将数据存储到数据包中
System.out.println("rtp UDP接收包===>"+datagramPacket.getSocketAddress());
handlerReceiveData(1,datagramPacket);
} catch (IOException e) {
e.printStackTrace();
break;
}
}
}
});
return rtcpUdpSocket.getLocalPort();
}
public static void handlerReceiveData(int type,DatagramPacket datagramPacket){
if(type==0){
// System.out.println("组播");
// System.out.println("接收到参数："+datagramPacket.getSocketAddress());
// System.out.println(datagramPacket.getData());
// System.out.println(datagramPacket.getLength());
}else{
System.out.println("rtcp");
// System.out.println("接收到参数："+datagramPacket.getSocketAddress());
// System.out.println(datagramPacket.getData());
// System.out.println(datagramPacket.getLength());
}
}
}
-
RTPH264Server.java
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
public class RTPH264Server {
private RandomAccessFile in;
private List<Long> NALUIndexs = new ArrayList<>() ;//用来记录每个NALU的起始位置
DatagramSocket rtpUdpSocket;//发送数据的socket
InetAddress clientAddress;//目标地址
int clientRtpPort;//目标端口
public RTPH264Server(DatagramSocket socket, InetAddress address, int port) throws SocketException {
rtpUdpSocket = socket;
clientAddress = address;
clientRtpPort = port;
}
public void startSendRtpPackage() throws Exception {
String fileName = RTPH264Server.class.getResource("test.h264").getPath();
in = new RandomAccessFile(fileName, "r");
parseIndexs();//获取所有起始下标
sendNALURtpPackage();
in.close();
}
/*
* 获取所有NAUL的起始位置
*/
public void parseIndexs() throws IOException {
while(true) {
if(in.length()>0&&parseNALU()>0) {
//parseNALU寻找NALU的起始位置(001或0001后面的位置)
NALUIndexs.add(in.getFilePointer());//getFilePointer()返回此文件中的当前偏移量。
}
if(in.length()-in.getFilePointer()<4) {
//读到文件尾部，跳出
break;
}
// System.out.println(in.getFilePointer());
// in.seek(in.getFilePointer()-4);//getFilePointer()返回此文件中的当前偏移量。
// System.out.println(in.getFilePointer());
// in.readByte();//从此文件中读取一个带符号的八位值。
// System.out.println(in.getFilePointer());
}
}
/*
* H.264原始码流：由多个NALU组成
* 每个NALU之间用起始码（0x000001(3Byte)或0x00000001(4Byte)）分割
* H.264编码时，在每个NAL前添加起始码0x000001，解码器在码流中检测到起始码，当前NAL结束；
* 为了防止NAL内部出现0x000001的数据，h.264又提出'防止竞争 emulation prevention"机制，
* 在编码完一个NAL时，如果检测出有连续两个0x00字节，就在后面插入一个0x03；
* 当解码器在NAL内部检测到 0x000003的数据，就把0x03抛弃，恢复原始数据
* */
public int parseNALU() throws IOException {
int head = in.readInt();//从该文件读取一个带符号的32位整数,一次读32位=4byte；0x00 00 00 01
if(head==1) {//0x00000001?
return 4;
}else if(head>>8 == 1) {//0x000001？
in.seek(in.getFilePointer()-1);//getFilePointer()返回此文件中的当前偏移量；seek()设置文件指针偏移，从该文件的开头测量，发生下一次读取或写入
return 3;
}
return -1;
}
/*
* 获取每一帧NALU 并存入集合
*/
public void sendNALURtpPackage() throws IOException, InterruptedException {
int framerate = 25;//framerate是帧率，每秒多少帧，每秒多少张图片。
int timestamp_increse = (int) (90000.0 / framerate);//码率=90000bits/s，一帧则用90000/10=9000bits表示
int PT = 96;//负载类型号96:h264
int packageSize = 1400;//最大负载长度
int seqNum = 1;//序列号
int ts_current = 0;//当前时间戳
boolean isEnd=false;
for(int i=0;i<NALUIndexs.size();) {
in.seek(NALUIndexs.get(i));//设置文件指针偏移
int len = 0;
if(i!=NALUIndexs.size()-1) {
len = (int) (NALUIndexs.get(i+1)-NALUIndexs.get(i));
}else {
//最后一个NALU
len = (int) (in.length() - NALUIndexs.get(i));
}
byte[] h264NALUArr=new byte[len];
in.read(h264NALUArr);
List<byte[]> bytes = h264DataToRtp(h264NALUArr, packageSize, PT, seqNum, ts_current, 0x88923423);
// System.out.println(bytes.size());
for(byte[] arr:bytes){
//2.创建数据报，包含响应的数据信息
DatagramPacket packet2=new DatagramPacket(arr, arr.length,clientAddress,clientRtpPort);
try {
rtpUdpSocket.send(packet2);//3.响应客户端
} catch (IOException e) {
e.printStackTrace();
isEnd=true;
break;
}
}
ts_current+=timestamp_increse;
seqNum += bytes.size();
Thread.sleep(30);
/*实现循环播放*/
if(i==NALUIndexs.size()-1){
i=0;
}else{
i++;
}
if(isEnd){
break;
}
}
}
/*
* H.264的RTP打包方式:
* 1.单NALU打包:一个RTP包中包含一个完整的NALU
* 2.聚合打包:对于较小的NALU，一个RTP包可包含多个完整的NALU
* 3.分片打包:对于较大的NALU，一个NALU可以分为多个RTP包发送
* - 在RTP载荷开始有两个字节的信息，然后再是NALU的内容
* - 第一个字节位（F1|R2|Type5）Type=28
* - 第二个字节位（S1|E1|Type5）S是否第一包；E是否最后一包；
* */
public List<byte[]> h264DataToRtp(byte[] h264NALUArr,int packageSize,int PT,int seq,int timestamp,int ssrc){
List<byte[]> res=new ArrayList();//需要发送的rtp包数据
int seqNum = seq;//包序列号
if(h264NALUArr.length<=packageSize){
byte[] rtpHeader = initRTPHeader(PT, seqNum, timestamp, ssrc);//12个字节的rtpHeader
//1.单NALU打包:一个RTP包中包含一个完整的NALU
byte[] rtpPackage=new byte[12+h264NALUArr.length];
System.arraycopy(rtpHeader,0,rtpPackage,0,12);//从源数组的第几位，复制到目标数组开始下标，n位
System.arraycopy(h264NALUArr,0,rtpPackage,12,h264NALUArr.length);//从源数组的第几位，复制到目标数组开始下标，n位
res.add(rtpPackage);
}else{
//3.分片打包:对于较大的NALU，一个NALU可以分为多个RTP包发送
/*
* 0 1 2
* 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* | FU indicator | FU header | FU payload ... |
* +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
* |F|NRI| Type |S|E|R| Type |
* +---------------+--------------+
*/
byte head=h264NALUArr[0];
int pktNum = (h264NALUArr.length-1)/packageSize; // 有几个完整的包
int endPktSize = (h264NALUArr.length-1)%packageSize; // 剩余不完整包的大小
int currentNum = 0;
while (currentNum <= pktNum){
if(currentNum<pktNum){
byte[] rtpHeader = initRTPHeader(PT, seqNum, timestamp, ssrc);//12个字节的rtpHeader
byte[] rtpPackage=new byte[12+2+packageSize];//(currentNum*packageSize+1,packageSize)
System.arraycopy(rtpHeader,0,rtpPackage,0,12);//从源数组的第几位，复制到目标数组开始下标，n位
rtpPackage[12]= (byte) (head & 0x60 |(byte) (28));
if(currentNum==0){
//第一包
rtpPackage[13]= (byte) (0x80 | ((byte) (head & 0x1f)));//|S=1|E=0|R=0|
}else if(currentNum==pktNum-1&&endPktSize==0){
//最后一包
rtpPackage[13]= (byte) (0x40 | ((byte) (head & 0x1f)));//|S=0|E=1|R=0|
}else{
//中间包
rtpPackage[13]= (byte) (head & 0x1f);//|S=0|E=0|R=0|
}
System.arraycopy(h264NALUArr,currentNum*packageSize+1,rtpPackage,14,packageSize);//从源数组的第几位，复制到目标数组开始下标，n位
res.add(rtpPackage);
seqNum+=1;
}else if(currentNum==pktNum&&endPktSize>0){
//最后一包
byte[] rtpHeader = initRTPHeader(PT, seqNum, timestamp, ssrc);//12个字节的rtpHeader
byte[] rtpPackage=new byte[12+2+endPktSize];//(currentNum*packageSize+1,endPktSize)
System.arraycopy(rtpHeader,0,rtpPackage,0,12);//从源数组的第几位，复制到目标数组开始下标，n位
rtpPackage[12]= (byte) (head & 0x60 |(byte) (28));
rtpPackage[13]= (byte) (0x40 | ((byte) (head & 0x1f)));//|S=0|E=1|R=0|
System.arraycopy(h264NALUArr,currentNum*packageSize+1,rtpPackage,14,endPktSize);//从源数组的第几位，复制到目标数组开始下标，n位
res.add(rtpPackage);
seqNum+=1;
}
currentNum++;
}
}
return res;
}
/*
* RTP报文格式：
* |===============================================================|
* | 0 | 1 | 2 | 3 |
* |===============|===============|===============|===============|
* |7 6 5 4 3 2 1 0|7 6 5 4 3 2 1 0|7 6 5 4 3 2 1 0|7 6 5 4 3 2 1 0|
* |===============|===============|===============================|
* |V2|P1|X1|CC4 |M1| PT7 | sequence number16 |
* |===============================================================|
* | timestamp时间戳 |
* |===============================================================|
* |同步信源(SSRC)标识符synchronization source (SSRC) identifier |
* |===============================================================|
* |特约信源(CSRC)标识符contributing source (CSRC) identifiers |
* | .... |
* |===============================================================|
*/
public byte[] initRTPHeader(int PT,int seq,int timestamp,int ssrc){
byte[] headerArr = new byte[12];//rtp固定头部有12个字节
//1.清空headerArr
for (int i = 0; i < headerArr.length; i++) { headerArr[i] = (byte) 0; }
//2.填充数据
//字节1:|V2|P1|X1|CC4 |
headerArr[0] = (byte) 0x80;//10000000==>V=1.0,P=0,X=0,CC=0000
//字节2:|M1| PT7 |
headerArr[1] = (byte)(PT & 0x7f);//01100000==>M=0,PT=1100000
//字节3-4:|sequence number16|：headerArr[2],headerArr[3]
System.arraycopy(intToBytes(seq,2),0,headerArr,2,2);//从源数组的第几位，复制到目标数组开始下标，n位
//字节5-8|timestamp时间戳|：headerArr[4]~headerArr[7]
System.arraycopy(intToBytes(timestamp,4),0,headerArr,4,4);//从源数组的第几位，复制到目标数组开始下标，n位
//字节9-12|同步信源(SSRC)标识符|：headerArr[8]~headerArr[11]
System.arraycopy(intToBytes(ssrc,4),0,headerArr,8,4);//从源数组的第几位，复制到目标数组开始下标，n位
return headerArr;
}
/**
* 将32位长度转换为n字节。（大端字节序：高位在前，低位在后）
* @param ldata 将从中构造n字节数组的int。
* @param n 要将长文件转换为的所需字节数。
* @return 用长值填充的所需字节数组。
*/
public byte[] intToBytes(int ldata, int n) {
byte[] buff = new byte[n];
for (int i=n-1;i>=0;i--) {
// 保持将最右边的8位分配给字节数组，同时在每次迭代中移位8位
buff[i] = (byte)ldata;
ldata = ldata>>8;
}
return buff;
}
}
启动
- 运行
RtspTcpServer.java
mian()
java MulticastSocket实现传输H.264的组播RTSP服务传输H.264的组播RTSP服务和java实现传输H.264的RTSP服务区别代码启动 - 使用vlc播放器播放网络流（文件=>打开网络=>输入URL）
rtsp://127.0.0.1:8888
 | | |