开发推流拉流工具类

Pu Zhibing

2025-04-09 b8a3a680f3e6720a8329bfaae571b09659aace52

开发推流拉流工具类

 ruoyi-api/ruoyi-api-dataInterchange/src/main/java/com/ruoyi/dataInterchange/api/vo/UPPlaybackMsgStartupAckVo.java

@@ -50,4 +50,8 @@
     * 0x05: 不满足跨域条件
     */
    private int result;
    /**
     * 播放地址
     */
    private String url;
}

 ruoyi-api/ruoyi-api-dataInterchange/src/main/java/com/ruoyi/dataInterchange/api/vo/UPRealvideoMsgStartupAckVo.java

@@ -50,4 +50,8 @@
     * 企业视频服务端口号
     */
    private int serverPort;
    /**
     * 播放地址
     */
    private String url;
}

 ruoyi-api/ruoyi-api-system/src/main/java/com/ruoyi/system/api/model/Car.java

@@ -260,4 +260,8 @@
    @TableField(exist = false)
    @ApiModelProperty(value = "驾驶时长")
    private Long drivingTime;
	
    @TableField(exist = false)
    @ApiModelProperty(value = "速度")
    private Integer speed;
}

 ruoyi-common/ruoyi-common-security/src/main/java/com/ruoyi/common/security/utils/SecurityUtils.java

@@ -106,7 +106,7 @@
    
    
    public static void main(String[] args) {
        String s = encryptPassword("0173940ec8314d1671c2ba365d28fcc1");
        String s = encryptPassword("0192023a7bbd73250516f069df18b500");
        System.err.println(s);

//        System.err.println(matchesPassword("c4ca4238a0b923820dcc509a6f75849b", "$2a$10$/7z/wiZ2ejTjD4GQuuyiSe1ptS8uoxrRAUlGCitiLEmpOO78cISeq"));

 ruoyi-gateway/src/main/resources/banner.txt

File was deleted

 ruoyi-modules/ruoyi-system/pom.xml

@@ -125,6 +125,27 @@
            <version>3.3.0</version>
            <type>pom</type>
        </dependency>
        <!--hutool-all-->
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.0.3</version>
        </dependency>
		
        <dependency>
            <groupId>org.bytedeco</groupId>
            <artifactId>javacv-platform</artifactId>
            <version>1.5.10</version>
        </dependency>
        <dependency>
            <groupId>org.bytedeco</groupId>
            <artifactId>ffmpeg</artifactId>
            <version>6.1.1-1.5.10</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>
    </dependencies>
    
    <build>

 ruoyi-modules/ruoyi-system/src/main/java/com/ruoyi/system/controller/CarController.java

@@ -20,6 +20,7 @@
import com.ruoyi.system.service.ICarTypeService;
import com.ruoyi.system.service.IDriverService;
import com.ruoyi.system.service.IEnterpriseService;
import com.ruoyi.system.util.JavaCVStreamUtil;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
@@ -34,6 +35,10 @@
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author zhibing.pu
@@ -43,30 +48,23 @@
@RequestMapping("/car")
public class CarController {
    
    private String outputUrl = "F:\\nginx 1.7.11.3 Gryphon\\html\\";
    @Resource
    private ICarService carService;
	
    @Resource
    private IDriverService driverService;
	
    @Resource
    private IEnterpriseService enterpriseService;
	
    @Resource
    private RealVideoMsgClient realVideoMsgClient;
	
    @Resource
    private PlaybackMsgClient playbackMsgClient;
	
    @Resource
    private UPExgMsgRealLocationClient upExgMsgRealLocationClient;
	
    @Resource
    private ICarTypeService carTypeService;
	
    @Resource
    private RedisTemplate redisTemplate;
	
    
    @GetMapping("/getCarList")
    @ApiOperation(value = "获取车辆列表", tags = {"车辆管理"})
@@ -113,12 +111,37 @@
        if (200 == msgStartupAckVoR.getCode()) {
            UPRealvideoMsgStartupAckVo data = msgStartupAckVoR.getData();
            RealVideoResp resp = new RealVideoResp();
            resp.setServerIp(data.getServerIP());
            resp.setServerPort(data.getServerPort());
            String url = "rtmp://192.168.110.85:1935/flv-live/" + car.getVehicleNumber();
            //执行拉流和推流
            ExecutorService executorService = new ThreadPoolExecutor(1, 1,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>());
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    JavaCVStreamUtil.push_flv(data.getUrl(), url, id);
                }
            });
            resp.setUrl(url);
            return R.ok(resp);
        }
		
        return R.fail(msgStartupAckVoR.getMsg());
    }
	
	
    @GetMapping("/closeRealVideo/{id}")
    @ApiOperation(value = "关闭实时音视频", tags = {"车辆管理"})
    @ApiImplicitParams({
            @ApiImplicitParam(value = "车辆id", name = "id", required = true)
    })
    public R closeRealVideo(@PathVariable("id") Integer id){
        JavaCVStreamUtil.close(id);
        return R.ok();
    }
	
	
	
    
    
    @GetMapping("/getPlaybackVideo")
@@ -134,8 +157,18 @@
        if (200 == startupAckVoR.getCode()) {
            UPPlaybackMsgStartupAckVo data = startupAckVoR.getData();
            RealVideoResp resp = new RealVideoResp();
            resp.setServerIp(data.getServerIP());
            resp.setServerPort(data.getServerPort());
            String url = "rtmp://192.168.110.85:1935/flv-live/" + car.getVehicleNumber();
            //执行拉流和推流
            ExecutorService executorService = new ThreadPoolExecutor(1, 1,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>());
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    JavaCVStreamUtil.push_flv(data.getUrl(), url, req.getId());
                }
            });
            resp.setUrl(url);
            return R.ok(resp);
        }
        return R.fail(startupAckVoR.getMsg());
@@ -223,6 +256,7 @@
            if (null != gnssDataVo) {
                car.setLongitude(new BigDecimal(gnssDataVo.getLon()).divide(new BigDecimal(1000000)).toString());
                car.setLatitude(new BigDecimal(gnssDataVo.getLat()).divide(new BigDecimal(1000000)).toString());
                car.setSpeed(gnssDataVo.getVec1());
            }
        }
        return R.ok(list);

 ruoyi-modules/ruoyi-system/src/main/java/com/ruoyi/system/controller/SysUserController.java

@@ -4,6 +4,7 @@
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.ruoyi.common.core.domain.R;
import com.ruoyi.common.core.utils.MD5Util;
import com.ruoyi.common.core.utils.StringUtils;
import com.ruoyi.common.core.web.controller.BaseController;
import com.ruoyi.common.core.web.domain.AjaxResult;
@@ -112,7 +113,7 @@
        if (StringUtils.isNotEmpty(user.getUserName()) && !userService.checkUserNameUnique(user)) {
            return error("登录账号重复");
        }
        user.setPassword(SecurityUtils.encryptPassword("123456"));
        user.setPassword(SecurityUtils.encryptPassword(MD5Util.getMD5("123456")));
        user.setDelFlag("0");
        user.setCreateTime(new Date());
        userService.save(user);

 ruoyi-modules/ruoyi-system/src/main/java/com/ruoyi/system/query/RealVideoResp.java

@@ -13,4 +13,6 @@
    private String serverIp;
    @ApiModelProperty("视频服务器端口号")
    private Integer serverPort;
    @ApiModelProperty("视频地址")
    private String url;
}

 ruoyi-modules/ruoyi-system/src/main/java/com/ruoyi/system/util/JavaCVStreamUtil.java

New file
@@ -0,0 +1,242 @@
package com.ruoyi.system.util;

import org.bytedeco.ffmpeg.global.avcodec;
import org.bytedeco.ffmpeg.global.avutil;
import org.bytedeco.javacv.*;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 音视频推流拉流工具类
 * 参考资料：https://juejin.cn/post/7311230172237561866
 * @author zhibing.pu
 * @Date 2025/4/8 16:03
 */
public class JavaCVStreamUtil {
	
    private static Map<Integer, FFmpegFrameGrabber> grabberMap = new ConcurrentHashMap<>();
	
    private static Map<Integer, FFmpegFrameRecorder> frameRecorderMap = new ConcurrentHashMap<>();
	
    private static Map<Integer, Boolean> statusMap = new ConcurrentHashMap<>();
	
	
	
    /**
     * 视频拉流和推流
     */
    public static void push_flv(String inputUrl, String outputUrl, Integer deviceNumber){
//        OpenCVFrameGrabber grabber = null;
        FFmpegFrameGrabber grabber = null;
        FFmpegFrameRecorder recorder = null;
        try {
            //关闭上一个没有正确关闭的流
            FFmpegFrameGrabber fFmpegFrameGrabber = grabberMap.get(deviceNumber);
            if(null != fFmpegFrameGrabber){
                fFmpegFrameGrabber.close();
            }
            FFmpegFrameRecorder fFmpegFrameRecorder = frameRecorderMap.get(deviceNumber);
            if(null != fFmpegFrameRecorder){
                fFmpegFrameRecorder.close();
            }
			
            //设置FFmpeg日志级别
            avutil.av_log_set_level(avutil.AV_LOG_INFO);
            FFmpegLogCallback.set();
			
            //视频抓帧
            grabber = new FFmpegFrameGrabber(inputUrl);
            grabber.setOption("rtsp_transport", "tcp");
            // 正确设置超时时间
            grabber.setOption("timeout", "120000");
            grabber.start();
            grabberMap.put(deviceNumber, grabber);
			
//            grabber = new OpenCVFrameGrabber(0);
//            grabber.start();
			
            //录制视频，推送到流媒体服务器（nginx）
            recorder = new FFmpegFrameRecorder(outputUrl, grabber.getImageWidth(), grabber.getImageHeight());
            recorder.setFormat("flv");
            // 设置视频比特率
            recorder.setVideoBitrate(grabber.getVideoBitrate());
//            // 设置帧率
//            recorder.setFrameRate(grabber.getVideoFrameRate());
//            // 设置关键帧间隔
//            recorder.setGopSize((int) grabber.getVideoFrameRate());
            // CRF 是一种用于控制视频/音频质量的参数，它允许在保持目标质量的同时动态地调整比特率。较低的CRF值表示更高的质量，但也可能导致较大的文件大小
            recorder.setAudioOption("crf", "23");
			
            //设置音频编码为AAC
//            if (grabber.getAudioChannels() > 0) {
                recorder.setAudioChannels(grabber.getAudioChannels());
                recorder.setAudioBitrate(grabber.getAudioBitrate());
                recorder.setAudioCodec(avcodec.AV_CODEC_ID_AAC);
//            }
            recorder.setVideoCodec(avcodec.AV_CODEC_ID_H264);
            //将解码后的帧记录到输出文件中
            //recorder.start通常用于处理已经解码成图像的视频数据
            recorder.start();
            Frame frame;
            while ((frame = grabber.grab()) != null) {
                recorder.record(frame);
            }
        } catch (FrameGrabber.Exception | FrameRecorder.Exception e) {
            e.printStackTrace();
        }finally {
            try {
                statusMap.put(deviceNumber, false);
                if(null != grabber){
                    grabber.stop();
                }
                if(null != recorder){
                    recorder.stop();
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
	
	
    /**
     * 视频拉流和推流
     */
    public static void push_hls(String inputUrl, String outputUrl, Integer deviceNumber){
        OpenCVFrameGrabber grabber = null;
//        FFmpegFrameGrabber grabber = null;
        FFmpegFrameRecorder recorder = null;
//        outputUrl = "http://192.168.110.85:80/test.m3u8";
        try {
            //关闭上一个没有正确关闭的流
            FFmpegFrameGrabber fFmpegFrameGrabber = grabberMap.get(deviceNumber);
            if(null != fFmpegFrameGrabber){
                fFmpegFrameGrabber.close();
            }
            FFmpegFrameRecorder fFmpegFrameRecorder = frameRecorderMap.get(deviceNumber);
            if(null != fFmpegFrameRecorder){
                fFmpegFrameRecorder.close();
            }
			
            //设置FFmpeg日志级别
            avutil.av_log_set_level(avutil.AV_LOG_INFO);
            FFmpegLogCallback.set();
			
            //视频抓帧
//            grabber = new FFmpegFrameGrabber(inputUrl);
//            grabber.setOption("rtsp_transport", "tcp");
//            // 正确设置超时时间
//            grabber.setOption("timeout", "30000");
//            grabber.start();
//            grabberMap.put(deviceNumber, grabber);
			
            grabber = new OpenCVFrameGrabber(0);
            grabber.start();
			
            //录制视频，推送到流媒体服务器（nginx）
            recorder = new FFmpegFrameRecorder(outputUrl, grabber.getImageWidth(), grabber.getImageHeight());
            recorder.setFormat("hls");
            // 设置视频比特率
            recorder.setVideoBitrate(grabber.getVideoBitrate());
            // 设置帧率
//            recorder.setFrameRate(grabber.getVideoFrameRate());
//            // 设置关键帧间隔
//            recorder.setGopSize((int) grabber.getVideoFrameRate());
			
            // 设置HLS切片参数
            //将每个切片时长设置为10秒
            recorder.setOption("hls_time", "15");
            //设置切片数大小
            recorder.setOption("hls_list_size", "20");
            //设置切片循环次数为50
            recorder.setOption("hls_wrap", "20");
            //每次切片完成后，都会删除之前的切片文件。如果不设置或设置为其他值，则不会删除之前的切片文件。
            recorder.setOption("hls_flags", "delete_segments");
            //在使用 H.264 编码时，通常要求输入的像素格式为 YUV420P。如果输入的像素格式不匹配，就可能导致 avcodec_send_frame() 错误
            recorder.setPixelFormat(avutil.AV_PIX_FMT_YUV420P);
            //CRF 是一种用于控制视频/音频质量的参数，它允许在保持目标质量的同时动态地调整比特率。较低的CRF值表示更高的质量，但也可能导致较大的文件大小
            recorder.setAudioOption("crf", "23");
			
            //设置音频编码为AAC
            if (grabber.getAudioChannels() > 0) {
                recorder.setAudioChannels(grabber.getAudioChannels());
                recorder.setAudioBitrate(grabber.getAudioBitrate());
                recorder.setAudioCodec(avcodec.AV_CODEC_ID_AAC);
            }
			
            recorder.setVideoCodec(avcodec.AV_CODEC_ID_H264);
            //设置音频编码为AAC
            if (grabber.getAudioChannels() > 0) {
                recorder.setAudioChannels(grabber.getAudioChannels());
                recorder.setAudioBitrate(grabber.getAudioBitrate());
                recorder.setAudioCodec(avcodec.AV_CODEC_ID_AAC);
            }
            //开始录制视频
            recorder.start();
            frameRecorderMap.put(deviceNumber, recorder);
            //设置状态为开始
            statusMap.put(deviceNumber, true);
			
            Frame frame;
            while ((frame = grabber.grab()) != null) {
                recorder.record(frame);
                //判断状态为停止，则结束此线程任务
                if(!statusMap.get(deviceNumber)){
                    break;
                }
            }
        } catch (FrameGrabber.Exception | FrameRecorder.Exception e) {
            e.printStackTrace();
        }finally {
            try {
                statusMap.put(deviceNumber, false);
                if(null != grabber){
                    grabber.stop();
                }
                if(null != recorder){
                    recorder.stop();
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
	
	
    /**
     * 关闭推流和拉流进程
     * @param deviceNumber
     */
    public static void close(Integer deviceNumber){
        //设置状态为停止
        statusMap.put(deviceNumber, false);
        FFmpegFrameGrabber fFmpegFrameGrabber = null;
        FFmpegFrameRecorder fFmpegFrameRecorder = null;
        try {
            fFmpegFrameGrabber = grabberMap.get(deviceNumber);
            if(null != fFmpegFrameGrabber){
                fFmpegFrameGrabber.close();
            }
            fFmpegFrameRecorder = frameRecorderMap.get(deviceNumber);
            if(null != fFmpegFrameRecorder){
                fFmpegFrameRecorder.close();
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try {
                fFmpegFrameGrabber = grabberMap.get(deviceNumber);
                if(null != fFmpegFrameGrabber){
                    fFmpegFrameGrabber.close();
                }
                fFmpegFrameRecorder = frameRecorderMap.get(deviceNumber);
                if(null != fFmpegFrameRecorder){
                    fFmpegFrameRecorder.close();
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}

 ruoyi-modules/ruoyi-system/src/main/java/com/ruoyi/system/util/websocket/MyWebSocketHandler.java

New file
@@ -0,0 +1,81 @@
package com.ruoyi.system.util.websocket;

import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.BinaryMessage;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.io.IOException;
import java.nio.ByteBuffer;

/**
 * @author zhibing.pu
 * @Date 2025/4/8 19:42
 */
@Component
@Slf4j
public class MyWebSocketHandler extends TextWebSocketHandler {
	
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        String sessionId = session.getId();
        log.info("WebSocket连接建立成功：{}", sessionId);
    }
	
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String payload = message.getPayload();
        log.info("收到消息：{}", payload);
        Integer carId = JSON.parseObject(payload).getInteger("carId");
        SessionMap.add(carId, session);
        // 发送回复消息
        String replyMessage = "服务器收到消息：" + payload;
        session.sendMessage(new TextMessage(replyMessage));
    }
	
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        String sessionId = session.getId();
        log.info("WebSocket连接关闭：{}", sessionId);
    }
	
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        log.error("WebSocket传输错误", exception);
    }
	
    // 广播消息给所有连接的客户端
    public void broadcastMessage(String message) {
        SessionMap.getAll().values().forEach(session -> {
            try {
                session.sendMessage(new TextMessage(message));
            } catch (IOException e) {
                log.error("广播消息失败", e);
            }
        });
    }
	
	
    /**
     * 发送消息
     * @param id
     * @param byteBuffer
     */
    public void send(Integer id, ByteBuffer byteBuffer){
        WebSocketSession webSocketSession = SessionMap.get(id);
        if(null != webSocketSession && webSocketSession.isOpen()){
            try {
                webSocketSession.sendMessage(new BinaryMessage(byteBuffer));
            } catch (IOException e) {
                log.error("发送失败", e);
            }
        }else if(null != webSocketSession && !webSocketSession.isOpen()){
            SessionMap.del(id);
        }
		
    }
}

 ruoyi-modules/ruoyi-system/src/main/java/com/ruoyi/system/util/websocket/SessionMap.java

New file
@@ -0,0 +1,35 @@
package com.ruoyi.system.util.websocket;

import org.springframework.web.socket.WebSocketSession;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author zhibing.pu
 * @Date 2025/4/8 20:04
 */
public class SessionMap{
	
	
    private static final Map<Integer, WebSocketSession> sessions = new ConcurrentHashMap<>();
	
    public static void add(Integer key, WebSocketSession value){
        sessions.put(key, value);
    }
	
	
    public static void del(Integer key){
        sessions.remove(key);
    }
	
	
    public static Map<Integer, WebSocketSession> getAll(){
        return sessions;
    }
	
	
    public static WebSocketSession get(Integer key){
        return sessions.get(key);
    }
}

 ruoyi-modules/ruoyi-system/src/main/java/com/ruoyi/system/util/websocket/WebSocketConfig.java

New file
@@ -0,0 +1,23 @@
package com.ruoyi.system.util.websocket;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(webSocketHandler(), "/websocket")
                .setAllowedOrigins("*"); // 允许跨域访问
    }
	
    @Bean
    public MyWebSocketHandler webSocketHandler() {
        // 使用自定义的WebSocket处理器
        return new MyWebSocketHandler();
    }
}

 ruoyi-service/ruoyi-dataInterchange/pom.xml

@@ -6,7 +6,7 @@
        <artifactId>ruoyi-service</artifactId>
        <version>3.6.2</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>Long
    <modelVersion>4.0.0</modelVersion>
    
    <artifactId>ruoyi-modules-dataInterchange</artifactId>
    

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/controller/PlaybackMsgController.java

@@ -15,7 +15,10 @@
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.time.LocalDateTime;
import java.time.ZoneOffset;

/**
 * @author zhibing.pu
@@ -48,13 +51,13 @@
    @PostMapping("/playbackMsgStartup")
    public R<UPPlaybackMsgStartupAckVo> playbackMsgStartup(@RequestParam("inferiorPlatformId") Integer inferiorPlatformId, @RequestParam("vehicleNo") String vehicleNo,
                                                           @RequestParam("startTime") Long startTime, @RequestParam("endTime") Long endTime) {
        R r = downPlaybackMsgStartupService.playbackMsgStartup(inferiorPlatformId, vehicleNo, startTime, endTime);
        R<String> r = downPlaybackMsgStartupService.playbackMsgStartup(inferiorPlatformId, vehicleNo, startTime, endTime);
        if (200 != r.getCode()) {
            return r;
            return R.fail(r.getMsg());
        }
        //发送成功，定时任务获取响应结果
        int num = 0;
        LocalDateTime now = LocalDateTime.now();
        Long now = LocalDateTime.now().toEpochSecond(ZoneOffset.ofHours(8));
        while (true) {
            UPPlaybackMsgStartupAck upPlaybackMsgStartupAck = upPlaybackMsgStartupAckDao.findByInferiorPlatformIdAndVehicleNoAndCreateTimeAfter(inferiorPlatformId, vehicleNo, now);
            if (null == upPlaybackMsgStartupAck) {
@@ -72,6 +75,13 @@
                    case 0:
                        UPPlaybackMsgStartupAckVo vo = new UPPlaybackMsgStartupAckVo();
                        BeanUtils.copyProperties(upPlaybackMsgStartupAck, vo);
                        String encode = null;
                        try {
                            encode = URLEncoder.encode(vehicleNo, "UTF-8");
                        } catch (UnsupportedEncodingException e) {
                            throw new RuntimeException(e);
                        }
                        vo.setUrl("http://" + vo.getServerIP() + ":" + vo.getServerPort() + "/" + encode + "." + vo.getVehicleColor() + ".1.0." + r.getData());
                        return R.ok(vo);
                    case 1:
                        return R.fail("失败");
@@ -112,7 +122,7 @@
        }
        //发送成功，定时任务获取响应结果
        int num = 0;
        LocalDateTime now = LocalDateTime.now();
        long now = LocalDateTime.now().toEpochSecond(ZoneOffset.ofHours(8));
        while (true) {
            UPPlaybackMsgControlAck upPlaybackMsgControlAck = upPlaybackMsgControlAckDao.findByInferiorPlatformIdAndVehicleNoAndCreateTimeAfter(inferiorPlatformId, vehicleNo, now);
            if (null == upPlaybackMsgControlAck) {

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/controller/RealVideoMsgController.java

@@ -15,7 +15,10 @@
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.time.LocalDateTime;
import java.time.ZoneOffset;

/**
 * 实时音视频控制器
@@ -48,13 +51,13 @@
     */
    @PostMapping("/startupRealVideo")
    public R<UPRealvideoMsgStartupAckVo> startupRealVideo(@RequestParam("inferiorPlatformId") Integer inferiorPlatformId, @RequestParam("vehicleNo") String vehicleNo) {
        R realVideo = downRealvideoMsgStartupService.startupRealVideo(inferiorPlatformId, vehicleNo);
        R<String> realVideo = downRealvideoMsgStartupService.startupRealVideo(inferiorPlatformId, vehicleNo);
        if (realVideo.getCode() != 200) {
            return realVideo;
            return R.fail(realVideo.getMsg());
        }
        //发送成功，定时任务获取响应结果
        int num = 0;
        LocalDateTime now = LocalDateTime.now();
        long now = LocalDateTime.now().toEpochSecond(ZoneOffset.ofHours(8));
        while (true) {
            UPRealvideoMsgStartupAck realvideoMsgStartupAck = upRealvideoMsgStartupAckDao.findByInferiorPlatformIdAndVehicleNoAndCreateTimeAfter(inferiorPlatformId, vehicleNo, now);
            if (null == realvideoMsgStartupAck) {
@@ -72,6 +75,13 @@
                    case 0:
                        UPRealvideoMsgStartupAckVo vo = new UPRealvideoMsgStartupAckVo();
                        BeanUtils.copyProperties(realvideoMsgStartupAck, vo);
                        String encode = null;
                        try {
                            encode = URLEncoder.encode(vehicleNo, "UTF-8");
                        } catch (UnsupportedEncodingException e) {
                            throw new RuntimeException(e);
                        }
                        vo.setUrl("http://" + vo.getServerIP() + ":" + vo.getServerPort() + "/" + encode + "." + vo.getVehicleColor() + ".1.0." + realVideo.getData());
                        return R.ok(vo);
                    case 1:
                        return R.fail("失败");
@@ -108,7 +118,7 @@
        }
        //发送成功，定时任务获取响应结果
        int num = 0;
        LocalDateTime now = LocalDateTime.now();
        Long now = LocalDateTime.now().toEpochSecond(ZoneOffset.ofHours(8));
        while (true) {
            UPRealvideoMsgEndAck upRealvideoMsgEndAck = upRealvideoMsgEndAckDao.findByInferiorPlatformIdAndVehicleNoAndCreateTimeAfter(inferiorPlatformId, vehicleNo, now);
            if (null == upRealvideoMsgEndAck) {

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/dao/UPAuthorizeMsgStartupDao.java

@@ -19,4 +19,6 @@
     * @return
     */
    UPAuthorizeMsgStartup findByInferiorPlatformIdOrderByCreateTimeDesc(Integer inferiorPlatformId);
	
	
}

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/dao/UPPlaybackMsgControlAckDao.java

@@ -4,8 +4,6 @@
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
import org.springframework.stereotype.Repository;

import java.time.LocalDateTime;

/**
 * @author zhibing.pu
 * @Date 2025/3/24 10:13
@@ -23,5 +21,5 @@
     * @return
     */
    UPPlaybackMsgControlAck findByInferiorPlatformIdAndVehicleNoAndCreateTimeAfter(Integer inferiorPlatformId,
                                                                                   String vehicleNo, LocalDateTime createTime);
                                                                                   String vehicleNo, Long createTime);
}

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/dao/UPPlaybackMsgStartupAckDao.java

@@ -4,8 +4,6 @@
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
import org.springframework.stereotype.Repository;

import java.time.LocalDateTime;

/**
 * @author zhibing.pu
 * @Date 2025/3/24 10:10
@@ -22,5 +20,5 @@
     * @return
     */
    UPPlaybackMsgStartupAck findByInferiorPlatformIdAndVehicleNoAndCreateTimeAfter(Integer inferiorPlatformId,
                                                                                   String vehicleNo, LocalDateTime createTime);
                                                                                   String vehicleNo, Long createTime);
}

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/dao/UPRealvideoMsgEndAckDao.java

@@ -4,8 +4,6 @@
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
import org.springframework.stereotype.Repository;

import java.time.LocalDateTime;

/**
 * @author zhibing.pu
 * @Date 2025/3/24 9:39
@@ -20,5 +18,5 @@
     * @return
     */
    UPRealvideoMsgEndAck findByInferiorPlatformIdAndVehicleNoAndCreateTimeAfter(Integer inferiorPlatformId,
                                                                                String vehicleNo, LocalDateTime createTime);
                                                                                String vehicleNo, Long createTime);
}

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/dao/UPRealvideoMsgStartupAckDao.java

@@ -4,8 +4,6 @@
import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
import org.springframework.stereotype.Repository;

import java.time.LocalDateTime;

/**
 * @author zhibing.pu
 * @Date 2025/3/24 9:16
@@ -21,7 +19,7 @@
     * @return
     */
    UPRealvideoMsgStartupAck findByInferiorPlatformIdAndVehicleNoAndCreateTimeAfter(Integer inferiorPlatformId,
                                                                                    String vehicleNo, LocalDateTime createTime);
                                                                                    String vehicleNo, Long createTime);
    
    
}

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/model/DOWNBaseMsgVehicleAdded.java

@@ -6,6 +6,8 @@
import io.netty.buffer.Unpooled;
import lombok.Data;

import java.io.UnsupportedEncodingException;

/**
 * 补报车辆静态信息请求
 *
@@ -41,7 +43,12 @@
     */
    public byte[] encode() {
        ByteBuf byteBuf = Unpooled.buffer(28);
        byte[] bytes1 = this.getVehicleNo().getBytes();
        byte[] bytes1 = new byte[0];
        try {
            bytes1 = this.getVehicleNo().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 21; i++) {
            if (i < bytes1.length) {
                byteBuf.writeByte(bytes1[i]);

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/model/DOWNExgMsgReportDriverInfo.java

@@ -5,6 +5,8 @@
import io.netty.buffer.Unpooled;
import lombok.Data;

import java.io.UnsupportedEncodingException;

/**
 * 上报驾驶员身份信息请求
 *
@@ -36,7 +38,12 @@
     */
    public byte[] encode() {
        ByteBuf byteBuf = Unpooled.buffer(28);
        byte[] bytes1 = this.getVehicleNo().getBytes();
        byte[] bytes1 = new byte[0];
        try {
            bytes1 = this.getVehicleNo().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 21; i++) {
            if (i < bytes1.length) {
                byteBuf.writeByte(bytes1[i]);

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/model/DOWNExgMsgTakeEwaybillReq.java

@@ -5,6 +5,8 @@
import io.netty.buffer.Unpooled;
import lombok.Data;

import java.io.UnsupportedEncodingException;

/**
 * 上报车辆电子运单请求
 *
@@ -36,7 +38,12 @@
     */
    public byte[] encode() {
        ByteBuf byteBuf = Unpooled.buffer(28);
        byte[] bytes1 = this.getVehicleNo().getBytes();
        byte[] bytes1 = new byte[0];
        try {
            bytes1 = this.getVehicleNo().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 21; i++) {
            if (i < bytes1.length) {
                byteBuf.writeByte(bytes1[i]);

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/model/DOWNPlaybackMsgControl.java

@@ -5,6 +5,8 @@
import io.netty.buffer.Unpooled;
import lombok.Data;

import java.io.UnsupportedEncodingException;

/**
 * 远程录像回放控制请求
 *
@@ -61,7 +63,12 @@
     */
    public byte[] encode() {
        ByteBuf byteBuf = Unpooled.buffer(38);
        byte[] bytes1 = this.getVehicleNo().getBytes();
        byte[] bytes1 = new byte[0];
        try {
            bytes1 = this.getVehicleNo().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 21; i++) {
            if (i < bytes1.length) {
                byteBuf.writeByte(bytes1[i]);

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/model/DOWNPlaybackMsgStartup.java

@@ -5,6 +5,8 @@
import io.netty.buffer.Unpooled;
import lombok.Data;

import java.io.UnsupportedEncodingException;

/**
 * 远程录像回放请求
 *
@@ -77,7 +79,12 @@
     */
    public byte[] encode() {
        ByteBuf byteBuf = Unpooled.buffer(148);
        byte[] bytes1 = this.getVehicleNo().getBytes();
        byte[] bytes1 = new byte[0];
        try {
            bytes1 = this.getVehicleNo().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 21; i++) {
            if (i < bytes1.length) {
                byteBuf.writeByte(bytes1[i]);
@@ -94,7 +101,12 @@
        byteBuf.writeChar(this.getMemType());
        byteBuf.writeLong(this.getPlaybackStartTime());
        byteBuf.writeLong(this.getPlaybackEndTime());
        byte[] bytes2 = this.getAuthorizeCode().getBytes();
        byte[] bytes2 = new byte[0];
        try {
            bytes2 = this.getAuthorizeCode().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 64; i++) {
            if (i < bytes2.length) {
                byteBuf.writeByte(bytes2[i]);
@@ -102,7 +114,12 @@
                byteBuf.writeByte(0x00);
            }
        }
        byte[] bytes3 = this.getGnssData().getBytes();
        byte[] bytes3 = new byte[0];
        try {
            bytes3 = this.getGnssData().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 36; i++) {
            if (i < bytes3.length) {
                byteBuf.writeByte(bytes3[i]);

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/model/DOWNRealvideoMsgEnd.java

@@ -5,6 +5,8 @@
import io.netty.buffer.Unpooled;
import lombok.Data;

import java.io.UnsupportedEncodingException;

/**
 * 主动请求停止实时音视频传输
 *
@@ -47,7 +49,12 @@
     */
    public byte[] encode() {
        ByteBuf byteBuf = Unpooled.buffer(30);
        byte[] bytes1 = this.getVehicleNo().getBytes();
        byte[] bytes1 = new byte[0];
        try {
            bytes1 = this.getVehicleNo().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 21; i++) {
            if (i < bytes1.length) {
                byteBuf.writeByte(bytes1[i]);

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/model/DOWNRealvideoMsgStartup.java

@@ -5,6 +5,8 @@
import io.netty.buffer.Unpooled;
import lombok.Data;

import java.io.UnsupportedEncodingException;

/**
 * 实时音视频请求
 *
@@ -55,7 +57,12 @@
     */
    public byte[] encode() {
        ByteBuf byteBuf = Unpooled.buffer(130);
        byte[] bytes1 = this.getVehicleNo().getBytes();
        byte[] bytes1 = new byte[0];
        try {
            bytes1 = this.getVehicleNo().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 21; i++) {
            if (i < bytes1.length) {
                byteBuf.writeByte(bytes1[i]);
@@ -68,7 +75,12 @@
        byteBuf.writeInt(this.getDataLength());
        byteBuf.writeChar(this.getChannelId());
        byteBuf.writeChar(this.getAvttemType());
        byte[] bytes2 = this.getAuthorizeCode().getBytes();
        byte[] bytes2 = new byte[0];
        try {
            bytes2 = this.getAuthorizeCode().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 64; i++) {
            if (i < bytes2.length) {
                byteBuf.writeByte(bytes2[i]);
@@ -76,7 +88,12 @@
                byteBuf.writeByte(0x00);
            }
        }
        byte[] bytes3 = this.getGnssData().getBytes();
        byte[] bytes3 = new byte[0];
        try {
            bytes3 = this.getGnssData().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 36; i++) {
            if (i < bytes3.length) {
                byteBuf.writeByte(bytes3[i]);

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/model/DOWNWarnMsgUrgeTodoReq.java

@@ -5,6 +5,7 @@
import io.netty.buffer.Unpooled;
import lombok.Data;

import java.io.UnsupportedEncodingException;
import java.time.LocalDateTime;
import java.time.ZoneOffset;

@@ -97,7 +98,12 @@
     */
    public byte[] encode() {
        ByteBuf byteBuf = Unpooled.buffer(120);
        byte[] bytes1 = this.getVehicleNo().getBytes();
        byte[] bytes1 = new byte[0];
        try {
            bytes1 = this.getVehicleNo().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 21; i++) {
            if (i < bytes1.length) {
                byteBuf.writeByte(bytes1[i]);
@@ -114,7 +120,12 @@
        byteBuf.writeInt(this.getSupervisionId());
        byteBuf.writeLong(this.getSupervisionEndTime());
        byteBuf.writeByte(this.getSupervisionLevel());
        byte[] bytes2 = this.getSupervisor().getBytes();
        byte[] bytes2 = new byte[0];
        try {
            bytes2 = this.getSupervisor().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 16; i++) {
            if (i < bytes2.length) {
                byteBuf.writeByte(bytes2[i]);
@@ -122,7 +133,12 @@
                byteBuf.writeByte(0x00);
            }
        }
        byte[] bytes3 = this.getSupervisorTel().getBytes();
        byte[] bytes3 = new byte[0];
        try {
            bytes3 = this.getSupervisorTel().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 20; i++) {
            if (i < bytes3.length) {
                byteBuf.writeByte(bytes3[i]);
@@ -130,7 +146,12 @@
                byteBuf.writeByte(0x00);
            }
        }
        byte[] bytes4 = this.getSupervisorEmail().getBytes();
        byte[] bytes4 = new byte[0];
        try {
            bytes4 = this.getSupervisorEmail().getBytes("GBK");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
        for (int i = 0; i < 32; i++) {
            if (i < bytes4.length) {
                byteBuf.writeByte(bytes4[i]);

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/server/DOWNPlaybackMsgStartupService.java

@@ -38,7 +38,7 @@
     * @param vehicleNo
     * @return
     */
    public R playbackMsgStartup(Integer inferiorPlatformId, String vehicleNo, Long startTime, Long endTime) {
    public R<String> playbackMsgStartup(Integer inferiorPlatformId, String vehicleNo, Long startTime, Long endTime) {
        Channel channel = ChannelMap.getClientChannel(inferiorPlatformId);
        if (!channel.isActive()) {
            return R.fail("失败");
@@ -66,6 +66,6 @@
        OuterPacket out = new OuterPacket(DataType.DOWN_PLAYBACK_MSG.getCode(), body);
        channel.writeAndFlush(out);
        log.info("远程录像回放请求({})：{}", DataType.DOWN_PLAYBACK_MSG_STARTUP.getCode(), JSON.toJSONString(downPlaybackMsgStartup));
        return R.ok();
        return R.ok(authorizeMsgStartup.getAuthorizeCode1());
    }
}

 ruoyi-service/ruoyi-dataInterchange/src/main/java/com/ruoyi/dataInterchange/server/DOWNRealvideoMsgStartupService.java

@@ -38,7 +38,7 @@
     *
     * @param vehicleNo
     */
    public R startupRealVideo(Integer inferiorPlatformId, String vehicleNo) {
    public R<String> startupRealVideo(Integer inferiorPlatformId, String vehicleNo) {
        Channel channel = ChannelMap.getClientChannel(inferiorPlatformId);
        if (!channel.isActive()) {
            return R.fail("失败");
@@ -47,7 +47,7 @@
        UPExgMsgRegister exgMsgRegister = upExgMsgRegisterDao.findByVehicleNo(vehicleNo);
        UPAuthorizeMsgStartup authorizeMsgStartup = upAuthorizeMsgStartupDao.findByInferiorPlatformIdOrderByCreateTimeDesc(inferiorPlatformId);
        if (null == authorizeMsgStartup) {
            return R.fail("获取失效口令失败");
            return R.fail("获取时效口令失败");
        }
        DOWNRealvideoMsgStartup downRealvideoMsgStartup = new DOWNRealvideoMsgStartup();
        downRealvideoMsgStartup.setVehicleNo(exgMsgRegister.getVehicleNo());
@@ -62,7 +62,7 @@
        OuterPacket out = new OuterPacket(DataType.DOWN_REALVIDEO_MSG.getCode(), body);
        channel.writeAndFlush(out);
        log.info("发起实时音视频请求({})：{}", DataType.DOWN_REALVIDEO_MSG_STARTUP.getCode(), JSON.toJSONString(downRealvideoMsgStartup));
        return R.ok();
        return R.ok(authorizeMsgStartup.getAuthorizeCode1());
    }
    
}