java采用WebSocket向前端传输rtp封装的h264码流，使用wfs.js接收h264流，并在前端页面播放_java rtp

作者：繁依Fanyi0 | 2024-04-09 10:18:20

踩

java rtp

使用：监听到项目启动后就开启Udpserver,然后再网页上选择下拉列表，开始播放（需要有一个udp一直发送rtp包，收到了之后通过WebSocket发送给前端即可）。
gitee下载

Maven依赖

导入webSocket的支持jar包

    <dependency>
      <groupId>javax</groupId>
      <artifactId>javaee-api</artifactId>
      <version>7.0</version>
      <scope>provided</scope>
    </dependency>
1
2
3
4
5
6

WebSocket后端代码

这里我写的是当请求视频时，前端才会与后端建立长连接，然后后端开始推送视频流送给前端解析。里面用HashMap存储每个session对象，并有唯一标识，方便发送视频流时调用。这里借鉴了后端的websocket入门代码。先用文件测试，能够发送成功之后开始移植发送rtp封装h264的视频流。

import java.io.*;
import java.nio.ByteBuffer;
import java.util.*;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

/**
 * @ServerEndpoint 注解是一个类层次的注解，它的功能主要是将目前的类定义成一个websocket服务器端,
 *                 注解的值将被用于监听用户连接的终端访问URL地址,客户端可以通过这个URL来连接到WebSocket服务器端
 */
@ServerEndpoint(value = "/{devid}")//{}中的数据代表一个参数，多个参数用/分隔
public class WebSocketTest {
    // 用来存放每个客户端对应的WebSocket对象
    public static final HashMap<String, WebSocketTest> dev_webSocket = new HashMap<>();
    // 与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;
    /**
     * 连接建立成功调用的方法
     *
     * @param session 可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    @OnOpen
    public void onOpen(@PathParam(value = "devid") String devid, Session session) {
        this.session = session;
        dev_webSocket.put(devid, this); // 加入map中
        System.out.println("有连接接入" + devid);
//        sendMessage(new File("xxx.h264"));先用文件测试是否能播放
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(@PathParam(value = "devid") String devid) {
        dev_webSocket.remove(devid); // 从map中删除
        System.out.println(devid + "连接关闭");
        timer.cancel();
        try {
            session.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发生错误时调用
     */
    @OnError
    public void onError(Session session, Throwable error) {
        System.out.println("发生错误" + error.getMessage());
    }

    /**
     * 这个方法与上面几个方法不一样。没有用注解，是根据自己需要添加的方法。
     */
    public void sendMessage(String message) {
        try {
            this.session.getBasicRemote().sendText(message);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void sendMessage(byte[] video) {
        System.out.println("本帧数据长度为"+video.length);
        try {
            ByteBuffer bf = ByteBuffer.wrap(video);
            this.session.getBasicRemote().sendBinary(bf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 这里以sps / pps / 帧 分成三段发送
     * 第一次sps 第二次 pps 第三次就是数据（分片需要拼成整体之后再送）周而复始
     * @param file
     */
    public void sendMessage(File file) {
        FileInputStream fileInputStream=null;
        try {
            fileInputStream= new FileInputStream(file);
            int len=fileInputStream.available();
            byte[] tmp=new byte[len];
            byte[] frame;
            ByteBuffer byteBuffer;
            //找到帧
            int front=0;
            int read = fileInputStream.read(tmp);
            System.out.println("读完为-1"+read);
            for (int i=0;i< len;i++){
                if (i+3>len){
                    return;
                }
                if (tmp[i]==0&&tmp[i+1]==0&&tmp[i+2]==0&&tmp[i+3]==1){//非一次开头
                    if (i-4<0){
                        continue;
                    }
                    frame=new byte[i-front];
                    System.arraycopy(tmp,front,frame,0,frame.length);
                    byteBuffer=ByteBuffer.wrap(frame);
                    this.session.getBasicRemote().sendBinary(byteBuffer);
//                    System.out.println("发送数据帧长度"+Arrays.toString(frame));
                    front=i;
                }
            }
        } catch (IOException e) {
            System.out.println("用户退出网页");
        } finally {
            if (fileInputStream!=null){
                try {
                    fileInputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

Udp接收rtp数据

开一个Udp的服务器，接收设备发送的rtp包，接收到之后，根据用户请求的唯一标识，将处理rtp包装的h264重新组包之后送给前端。标识现在默认为admin

package com;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


public class UdpReceiveData extends Thread{
    private DatagramSocket server;
    public static ThreadPoolExecutor threadPoolExecutor=new ThreadPoolExecutor(20,40,30,TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(10),new ThreadPoolExecutor.DiscardPolicy());
    public UdpReceiveData(int rtpPort) {
        try {
            this.server=new DatagramSocket(rtpPort);
        } catch (SocketException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        openRtp();
    }
    public void startRun(){
        threadPoolExecutor.execute(this);
    }
    /**
     * RTP的接收UDP  一般情况PT=96 为H264
     */
    public void openRtp() {
        System.out.println("**********************rtp开始接收数据**********************");
        DatagramPacket client;
        //一次接收数据的字节数组大小
        byte[] bytes=new byte[1500];
        RtpDataDeal rtpDataDeal=new RtpDataDeal();
        try {
            //开始收数据
            while (true) {
                //一次最大能接受2M 且每次需要重新定义否则，下一次的数据长度没这次长时，会将这次的数据填补
                client=new DatagramPacket(bytes, bytes.length);
                server.receive(client);
                //本次传输数据的长度
                WebSocketTest admin = WebSocketTest.dev_webSocket.get("admin");
                if (admin!=null){
                    byte[] bytes1 = rtpDataDeal.convergeBytes(client.getData(), client.getLength());
                    if (bytes1!=null){
                        admin.sendMessage(bytes1);
                    }
                }
            }
        } catch (IOException ignored) {
        }finally {//关闭
            server.close();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

处理组装rtp携带h264类

去掉rtp头后，不分片的直接在前面加上0001，分片的头部除了要去掉rtp头外，还要再减去rtp头部后两个字节，然后加上0001和rtp头部后两个字节的合并的一个字节，分片的其余部分直接去掉rtp头部和rtp头部的后两个字节，然后装在前面部分的后面即可。

package com;


import java.util.ArrayList;
import java.util.ListIterator;

public class RtpDataDeal extends Thread{
    //单个文件的标志
    private boolean singleFlag = false;
    //分片时nal的头字节由FU indicator的前三位和FU Header的后五位组成
    private byte dataHead;
    private final byte[] pierce = new byte[]{0, 0, 0, 1};//固定分割0001
    private final byte[] pierceAndHead=new byte[]{0,0,0,1,-1};//固定分割0001+两个字节组合成的头
    private ArrayList<byte[]> arrayList=new ArrayList<>(128);
    private int pierceLength=0;
    //分片的起始标志
    private boolean startFlag;
    private boolean endFlag;
    //rtp头部长度,也是除开rtp头后第一个字节的下标
    private static final int rtpHeadLength=12;
    /**
     * 解包时，取FU indicator的前三位和FU Header的后五位构成一个字节
     * 功能：分析处理rtp中h.264的的indicator和header
     *
     * @param fu1 头部的后两个字节(1)
     * @param fu2 头部的后两个字节(2)
     */
    public void dealNalData(byte fu1, byte fu2) {
        //FU indicator
        //f 0表示正常，128表示错误
        int f = fu1 & 0x80;//128   0
        //NRI 重要级别，11表示非常重要
        int nri = fu1 & 0x60;
        //FU Type 表示该NALU的类型是什么 28表示FU-A分片单元 1代表不分区 7代表SPS 8代表PPS 5代表IDR
        int fu_type = fu1 & 0x1f;//为28代表分包
        //FU Header
        //起始帧，为1表示分片的第一包
        //分包的起始包
        startFlag = (128 == (fu2 & 0x80));
        //末尾帧，为1表示分片的最后一包
        //结束包
        endFlag = (64 == (fu2 & 0x40));//前面表示要分片
        //表示不分片，一次单个NAL单元
        singleFlag = fu_type <= 23 && fu_type >= 1;
        //nal类型，表示为 什么帧
        int nalType = fu2 & 0x1f;
        if (fu_type == 28) {//分包时需将FU indicator的前三位和FU Header的后五位为1个字节放入
            dataHead = (byte) (f + nri + nalType);
        }
    }


    /**
     * 功能：拼接头部的4个字节 0 0 0 1 以及 nal 和 本次的荷载数据
     * @param rtpBody 本次的数据数组去掉rtp头之后的数据
     *  返还拼接了0 0 0 1 、 nal 和 荷载数据的数组
     */
    public byte[] convergeBytes(byte[] rtpBody,int length){
        dealNalData(rtpBody[rtpHeadLength],rtpBody[rtpHeadLength+1]);//处理此次的头两字节信息
        if (singleFlag) {//不分片
            //在头部加上0 0 0 1四个字节
            byte[] newByte=new byte[4+length-12];
            System.arraycopy(pierce,0,newByte,0,4);
            System.arraycopy(rtpBody,12,newByte,4,length-12);
            return newByte;
        }else if (startFlag){//要分片,分片的第一片才加，其余不加
            //+4 是加 0 0 0 1四个字节 +1 是加合并的头  -2 是减去头部的两个字节（因为这两个要合并成一个） +4 +1 -2
            byte[] newByte=new byte[3+length-12];
            pierceAndHead[4]=dataHead;
            System.arraycopy(pierceAndHead,0,newByte,0,5);
            //在头部加上0 0 0 1四个字节 和 dataHead 取indicator前3 和 head后5
            System.arraycopy(rtpBody,14,newByte,5,length-14);
            arrayList.add(newByte);
            pierceLength+=newByte.length;
            //这里因为rtpBody是整个数据包括前两个需要合并的字节，所以需要从rtpBody的第三个下标也就是2开始复制
            //因为从第三个字节开始复制，长度也需要减去2，因为不减长度的话没这么多位
            return null;
        }else {
            byte[] newByte=new byte[length-12-2];
            System.arraycopy(rtpBody,14,newByte,0,length-14);
            arrayList.add(newByte);
            pierceLength+=newByte.length;
            if (endFlag){
                return pinJie();
            }else return null;
        }
    }
    private byte[] pinJie(){
        byte[] needSend=new byte[pierceLength];
        ListIterator<byte[]> listIterator = arrayList.listIterator();
        int index=0;
        while (listIterator.hasNext()){
            byte[] next = listIterator.next();
            System.arraycopy(next,0,needSend,index,next.length);
            index+= next.length;
        }
        arrayList=new ArrayList<>(128);
        pierceLength=0;
        return needSend;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102

前端代码

<!DOCTYPE html>
<html lang="en">
	<head>
		<title>播放h264</title>
		<meta charset="utf-8">
		<script type="text/javascript" src="wfs.js"></script>
	</head>
	<body>
		<select id="sel" name="test">
			<option value="请选择">请选择</option>
			<option value="admin">admin</option>
		</select>
		<h2>播放h264</h2>
		<div class="wfsjs">
			<video muted id="video1" width="640" height="480" controls autoplay></video>
			<div class="ratio"></div>
		</div>
		<script src="jquery-3.4.1.js" type="text/javascript"></script>
		<script type="text/javascript">
			$(function () {
					$("#sel").on("change",function () {
						if ($("#sel").val()!="请选择"){
							if (Wfs.isSupported()) {
								var video1 = document.getElementById("video1"),wfss = new Wfs();
								// wfss.attachMedia(video1,'ch1',"H264Raw",$("#sel").val());
								wfss.attachMedia(video1,'ch1',"H264Raw","admin");//第四个即为设备标识，有需求可改为动态的
							}
						}
					})
			})
		</script>
	</body>
</html>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

引入的js文件

jquery-3.4.1.js
wfs.js（var client = new WebSocket(‘ws://’ + ‘localhost:8080’ + ‘/web/’+data.websocketName);localhost:8080/web为我tomcat的url,如果需要放到公网，需要修改localhost为公网ip）

待解决问题

经测试，功能基本实现，但存在以下问题：

无法自动播放，需要人为的点一下播放按钮，在火狐上，还需拖动一下进度条。
由于调用的wfs.js，不能直接使用close关闭websocket，刷新页面能关掉。
视频乱序，采用的udp传输有一定的乱序，目前我想到的方法就是，存一定的视频帧之后就开始排序，然后将前面的输出，缺失的就直接丢弃，借用了一点Tcp中滑动窗口的模型。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/繁依Fanyi0/article/detail/392053