代码运行报错：

Module 'GstDemo' production: java.lang.IllegalArgumentException: No enum constant java.lang.annotation.RetentionPolicy.ôÚ
Ìü
Ú
Ú
 £ÙÞÞ £É^ÂyÙÉ{  úÚ
Ú
 Ê: No enum constant java.lang.annotation.RetentionPolicy.ôÚ
Ìü
Ú
Ú
 £ÙÞÞ £É^ÂyÙÉ{  úÚ
Ú
 Ê

解决方式：

尝试卸载jdk，重装jdk，没有效果；

清除idea缓存，重启idea，解决。但是代码问题还是需要分析，这种方式治标不治本。

我的测试代码。

package org.example.appsrchope;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.example.gstreamer.X264EncoderConfig;
import org.example.utils.LoggingInitializer;
import org.freedesktop.gstreamer.*;
import org.freedesktop.gstreamer.elements.AppSrc;
import org.freedesktop.gstreamer.elements.PlayBin;
import org.freedesktop.gstreamer.glib.GLib;
import org.freedesktop.gstreamer.message.*;

import java.io.*;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.time.Instant;
import java.time.LocalTime;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;


/**
 * 视频处理工具类，支持视频转换和RTSP流推送
 */
@Slf4j
public class AppSrcUtil {

    static VideoLocalConfig localConfig = new VideoLocalConfig();

    private static final X264EncoderConfig DEFAULT_ENCODER_CONFIG = new X264EncoderConfig();


    // 控制推流线程的运行状态
    private static final AtomicBoolean isStreaming = new AtomicBoolean(false);

    public static void main(String[] args) throws InterruptedException, IOException {

        AppSrcUtil.init();

        System.out.println("Starting GStreamer push stream example...");

        // 创建管道输入输出流（实际应用中输出流可能来自其他音频源）
        PipedInputStream inputStream = new PipedInputStream();
        PipedOutputStream outputStream = new PipedOutputStream(inputStream);

        String videoRtspUrl = localConfig.getVideoRtspUrl();
        System.out.println("Starting video pipe push stream to RTSP: " + videoRtspUrl);
        isStreaming.set(true);

        // 启动音频写入线程（模拟音频数据输入）
        Thread videoWriterThread = new Thread(() -> {
            writeVideoData(outputStream);
        });
        videoWriterThread.start();

        // 启动音频推流线程
        Thread videoPushThread = new Thread(() -> {
            try {
                // 假设音频格式为AAC，根据实际情况调整
                boolean result = AppSrcUtil.pushStream2Rtsp(inputStream, videoRtspUrl);
                System.out.println("video pipe push stream result: " + (result ? "success" : "failed"));
            } finally {
                isStreaming.set(false);
                try {
                    inputStream.close();
                } catch (IOException e) {
                    System.err.println("Failed to close video pipe: " + e.getMessage());
                }
            }
        });
        videoPushThread.start();

        // 等待音频推流完成
        try {
            videoPushThread.join();
            videoWriterThread.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        System.out.println("All streaming examples completed.");
        System.exit(0);
    }

    @SneakyThrows
    private static void writeVideoData(PipedOutputStream output) {
        String tsFilePath = localConfig.getVideoFile();
        byte[] videoData;

        // 1️⃣ 一次性加载整个TS文件到内存
        try {
            File file = new File(tsFilePath);
            if (!file.exists()) throw new FileNotFoundException(tsFilePath);
            if (file.length() > 2L * 1024 * 1024 * 1024) { // >2GB
                throw new IllegalArgumentException("File too large");
            }
            videoData = Files.readAllBytes(file.toPath());
            log.info("Loaded TS file: {} bytes", videoData.length);
        } catch (Exception e) {
            log.error("Failed to load video file", e);
            isStreaming.set(false);
            output.close();
            return;
        }

        // 2️⃣ 配置推送参数（无需知道时长！）
        final int CHUNK_SIZE = 4096;
        // ⭐ 关键：按典型视频码率反推 sleep 时间
        // 假设码率范围：1 Mbps ~ 6 Mbps（常见TS流）
        // 4KB = 32768 bits
        // 在 2 Mbps 码率下，4KB 应耗时 ≈ 32768 / 2_000_000 ≈ 16 ms
        // 为安全起见，取稍慢速度（如 1.5 Mbps 等效）
        final long SLEEP_PER_CHUNK_MS = 25; // 经验值：适用于 1~4 Mbps 视频

        try {
            while (isStreaming.get()) {
                for (int offset = 0; offset < videoData.length && isStreaming.get(); ) {
                    int len = Math.min(CHUNK_SIZE, videoData.length - offset);
                    output.write(videoData, offset, len);
                    output.flush();
                    offset += len;

                    // 控制推送节奏，防止数据洪泛
                    try {
                        TimeUnit.MILLISECONDS.sleep(SLEEP_PER_CHUNK_MS);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return;
                    }
                }

                // 文件推完，短暂等待后循环（避免瞬间重推）
                if (isStreaming.get()) {
                    TimeUnit.MILLISECONDS.sleep(200); // 200ms gap between loops
                }
            }
        } catch (IOException e) {
            log.warn("Pipe write interrupted", e);
        } finally {
            try { output.close(); } catch (IOException ignored) {}
        }
    }


    static {
        LoggingInitializer.initLogging();
       // AppSrcUtil.init();
    }

    public static void init() {
        GLib.setEnv("GST_DEBUG", "*:5", true);

        // 初始化 GStreamer
        try { // 记录开始时间
            long initStartTime = System.currentTimeMillis();
            Gst.init("GStreamer Video Converter");
            long initEndTime = System.currentTimeMillis();
            log.info("GStreamer initialized successfully, cost = {} ms", initEndTime - initStartTime);
        } catch (Exception e) {
            log.error("Failed to initialize GStreamer: {}", e.getMessage());
            throw new RuntimeException("GStreamer initialization failed", e);
        }

    }


    /**
     * 用gstreamer将视频流推送到RTSP服务
     *
     * @param rtspUrl   RTSP服务地址
     */
    public static boolean pushStream2Rtsp(PipedInputStream inputStream , String rtspUrl) {
        Pipeline pipeline = null;
        CountDownLatch completionLatch = new CountDownLatch(1);
        AtomicBoolean success = new AtomicBoolean(false);

        // 记录开始时间
        long startTime = System.currentTimeMillis();
        String timestamp = LocalTime.now().toString();
        System.out.println(timestamp + "  开始加载文件");
        long loadTime = 0;

        try {
            // 创建管道
            pipeline = new Pipeline("rtsp-streaming-pipeline");

            // 使用 ElementFactory 创建元素
            AppSrc appsrc = (AppSrc) createAndCheckElement("appsrc", "video-appsrc");
            Element tsdemux = createAndCheckElement("tsdemux", "ts-demux");
            Element queue = createAndCheckElement("queue", "queue");
            Element h264parse = createAndCheckElement("h264parse", "h264parse");
            Element rtspclientsink = createAndCheckElement("rtspclientsink", "rtspclientsink");

            // 设置文件源路径
            appsrc.set("block", true);
            appsrc.setCaps(Caps.fromString("video/mpegts, systemstream=(boolean)true"));
            appsrc.set("format", Format.TIME);
            appsrc.set("stream-type", AppSrc.StreamType.STREAM);
            appsrc.set("do-timestamp", true);
            appsrc.set("is-live", true);

            // queue新增参数配置
            queue.set("max-size-time", 0);

            h264parse.set("config-interval", 1);

            // 设置 RTSP 目标地址
            rtspclientsink.set("location", rtspUrl);
            rtspclientsink.setAsString("protocols", "tcp");

            // 添加所有元素到管道
            pipeline.addMany(appsrc, tsdemux, queue, h264parse, rtspclientsink);

            // 链接基础元素
            if (!Element.linkMany(appsrc, tsdemux)) {
                log.error("Failed to link appsrc -> tsdemux");
                return false;
            }

            if (!Element.linkMany(h264parse, rtspclientsink)) {
                log.error("Failed to link queue to h264parse to rtspclientsink");
                return false;
            }

            // 添加总线监听器处理消息
            pipeline.getBus().connect((Bus.MESSAGE) (bus, message) -> {
                handlePushRtspStreamingMessage(message, completionLatch, success);
            });

            tsdemux.connect((Element.PAD_ADDED) (element, pad) -> {
                if (pad.getName().startsWith("video")) {
                    // 动态创建从 tsdemux 到 h264parse 的完整链路
                    try {
                        pad.link(queue.getStaticPad("sink")); // 先连到 queue
                        queue.getStaticPad("src").link(h264parse.getStaticPad("sink")); // 再连 queue → h264parse
                        log.info("Linked tsdemux → queue → h264parse");
                    } catch (Exception e) {
                        log.error("Failed to link demuxed video pad", e);
                        completionLatch.countDown();
                        isStreaming.set(false);
                    }
                }
            });


            // 添加信号监听以推送数据
            appsrc.connect((AppSrc.NEED_DATA) (elem, length) -> {
                try {
                    byte[] buffer = new byte[Math.min(length, 4096)];
                    int bytesRead = inputStream.read(buffer);
                    if (bytesRead > 0) {
                        Buffer gstBuffer = new Buffer(bytesRead);
                        gstBuffer.map(true).put(buffer, 0, bytesRead);
                        gstBuffer.unmap();
                        appsrc.pushBuffer(gstBuffer);
                    } else {
                        appsrc.endOfStream();
                        log.info("appsrc.endOfStream, push stream completed.");
                        completionLatch.countDown();
                        isStreaming.set(false);
                    }
                } catch (IOException e) {
                    log.error("Failed to read from audio pipe", e);
                    appsrc.endOfStream();
                    completionLatch.countDown();
                    isStreaming.set(false);
                }
            });

            // 启动管道
            pipeline.play();
            log.info("pipeline.play");

            loadTime = System.currentTimeMillis();
            timestamp = LocalTime.now().toString();
            log.info(timestamp + "  start to push streaming");


            // 等待推流完成（最多等待30分钟）
            if (!completionLatch.await(DEFAULT_ENCODER_CONFIG.getPushStream2RtspTimeout(), TimeUnit.SECONDS)) {
                log.warn("push stream timed out after {} seconds.", DEFAULT_ENCODER_CONFIG.getPushStream2RtspTimeout());
                return false;
            }

            if (success.get()) {
                log.info("push stream successfully.");
            } else {
                log.error("push stream failed.");
            }
            return success.get();
        } catch (Exception e) {
            log.error("push video [{}] to rtsp: {} failed, error: {}", localConfig.getVideoFile(), rtspUrl, e.getMessage());
            isStreaming.set(false);
            return false;
        } finally {
            // 计算并记录转换时间
            long endTime = System.currentTimeMillis();
            long pushDurationMillis = endTime - loadTime;
            double pushDurationSeconds = pushDurationMillis / 1000.0;
            long loadDurationMillis = loadTime - startTime;

            log.info("load video in {} ms, push stream in {} s", loadDurationMillis, String.format("%.2f", pushDurationSeconds));

            isStreaming.set(false);
            // 清理资源
            cleanupPipeline(pipeline);
        }
    }


    /**
     * 创建并检查元素
     */
    private static Element createAndCheckElement(String factoryName, String elementName) {
        Element element = ElementFactory.make(factoryName, elementName);
        if (element == null) {
            throw new RuntimeException("无法创建GStreamer元素: " + factoryName);
        }
        return element;
    }

    /**
     * 处理GStreamer消息（用于RTSP流推送）
     */
    private static void handlePushRtspStreamingMessage(Message message, CountDownLatch latch, AtomicBoolean success) {
        //log.info(LocalTime.now() + " Received message: {}", message.getType());
        if (message.getType() == MessageType.ERROR) {
            // 使用通用的方式获取错误信息
            Structure structure = message.getStructure();
            log.error("error message received, source: {}, structure: {}", message.getSource().getName(), structure);
            success.set(false);
            latch.countDown();
        } else if (message.getType() == MessageType.WARNING) {
            log.warn("Warning message received, source: {}, structure: {}", message.getSource().getName(), message.getStructure());
        } else if (message.getType() == MessageType.EOS) {
            log.info("End of stream reached, source: {}, structure: {}", message.getSource().getName(), message.getStructure());
            success.set(true);
            latch.countDown(); // 通知推流完成
        } else if (message.getType() == MessageType.STATE_CHANGED) {
            LocalTime now = LocalTime.now();
            StateChangedMessage scm = (StateChangedMessage) message;
            //if (scm.getSource().getName().startsWith("pipeline")) {
            State newState = scm.getNewState();
            State oldState = scm.getOldState();
            if (newState == State.PLAYING && oldState == State.PAUSED) {
                log.info("{} {} state changes from {} to {} ", now, scm.getSource().getName(), oldState, newState);
            }

        }
    }


    /**
     * 清理管道资源
     */
    private static void cleanupPipeline(Pipeline pipeline) {
        if (pipeline != null) {
            pipeline.stop();
            String timestamp = LocalTime.now().toString();
            log.info(timestamp + "  Pipeline stopped");
        }
    }


}

🔥 核心问题：多线程并发操作 GStreamer Pipeline + Buffer 生命周期管理错误

你的代码存在 两个严重隐患，会导致 native 层内存破坏（正是乱码异常的根源）：

---

❌ 问题 1：在多个线程中直接操作 GStreamer 对象（非线程安全）

📍 关键代码：

1// 主线程启动推流线程
2Thread videoPushThread = new Thread(() -> {
3    boolean result = AppSrcUtil.pushStream2Rtsp(inputStream, videoRtspUrl); // 👈 创建并操作 Pipeline
4});
5
6// 同时，main 线程可能还在做其他事（如接收 HTTP 请求等）

但更严重的是：

📍 在 pushStream2Rtsp 中：

1appsrc.connect((AppSrc.NEED_DATA) (elem, length) -> {
2    // 这个回调由 GStreamer 的 native 线程触发！
3    byte[] buffer = new byte[Math.min(length, 4096)];
4    int bytesRead = inputStream.read(buffer); // 👈 从 PipedInputStream 读取
5    if (bytesRead > 0) {
6        Buffer gstBuffer = new Buffer(bytesRead);
7        gstBuffer.map(true).put(buffer, 0, bytesRead);
8        gstBuffer.unmap();
9        appsrc.pushBuffer(gstBuffer); // 👈 在 GStreamer 线程中调用 pushBuffer
10    }
11});

✅ 这本身是合法的 —— NEED_DATA 回调允许调用 pushBuffer。

BUT！ 你在 writeVideoData 线程中：

1Thread videoWriterThread = new Thread(() -> {
2    writeVideoData(outputStream); // 👈 持续写入 PipedOutputStream
3});

这就形成了：

• GStreamer native 线程：从 PipedInputStream 读数据；
• Java 写入线程：向 PipedOutputStream 写数据；
• 主线程：可能调用 isStreaming.set(false) 或其他操作。

📌 风险点：
如果 inputStream 在 GStreamer 读取时被关闭（如因异常或超时），而 native 层仍在访问底层缓冲区，就会导致 use-after-free 或 buffer overflow → JVM 内存被破坏 → 乱码异常。

---

❌ 问题 2：Buffer 对象未正确管理生命周期（高危！）

📍 关键代码：

1Buffer gstBuffer = new Buffer(bytesRead);
2gstBuffer.map(true).put(buffer, 0, bytesRead);
3gstBuffer.unmap();
4appsrc.pushBuffer(gstBuffer);

⚠️ 这是错误的用法！

根据 gst1-java-core 文档 和 GStreamer 设计：

• Buffer 是 native 资源，一旦调用 pushBuffer()，所有权就移交给了 GStreamer；
• 你不应该再访问它，也不需要手动释放；
• 但你的代码中，gstBuffer 是局部变量，可能被 JVM 提前回收或 native 层重复释放。

更安全的做法是：让 GStreamer 完全控制 Buffer 生命周期。

---

✅ 正确修复方案

✅ 修复 1：确保 PipedInputStream 的关闭是线程安全的

在 pushStream2Rtsp 的 finally 块中关闭流，而不是在写入线程中随意关闭：

1} finally {
2    isStreaming.set(false);
3    try {
4        if (inputStream != null) {
5            inputStream.close(); // 由推流线程统一关闭
6        }
7    } catch (IOException e) {
8        log.warn("Failed to close input stream", e);
9    }
10    cleanupPipeline(pipeline);
11}

同时，在 writeVideoData 中捕获 IOException 并退出，不要主动关闭流：

1} catch (IOException e) {
2    log.warn("Pipe write interrupted", e);
3    // 不要在这里 close(output)，由推流线程统一处理
4    return;
5} finally {
6    // 移除 output.close()
7}

---

✅ 修复 2：避免主线程干扰（可选但推荐）

将整个推流逻辑封装到一个独立方法，并在单一专用线程中执行：

1public static void startStreaming(String videoPath, String rtspUrl) {
2    new Thread(() -> {
3        try {
4            // 创建 pipe
5            PipedInputStream in = new PipedInputStream();
6            PipedOutputStream out = new PipedOutputStream(in);
7            
8            // 启动写入线程
9            Thread writer = new Thread(() -> writeVideoData(out, videoPath));
10            writer.start();
11            
12            // 执行推流（阻塞直到完成）
13            pushStream2Rtsp(in, rtspUrl);
14            
15            writer.join();
16        } catch (Exception e) {
17            log.error("Streaming failed", e);
18        }
19    }, "GstStreamingThread").start();
20}

---

📌 总结：为什么会出现乱码异常？

原因	说明
native 内存越界	GStreamer native 层访问了已被 Java 关闭的 PipedInputStream 缓冲区
Buffer 生命周期错乱	Buffer 对象在 native 和 Java 之间双重管理，导致重复释放
多线程竞争	写入线程、GStreamer 线程、主线程同时操作共享资源

✅ 修复后效果：

• 不再出现 RetentionPolicy.ôÚÌü 这类乱码异常；
• 推流稳定性大幅提升；
• 即使循环推送也不会崩溃。

---

🔧 最终建议修改点（最小改动）

1. 移除 writeVideoData 中的 output.close()；
2. 只在 pushStream2Rtsp 的 finally 块中关闭 inputStream；