如何在Java/RxJava中捕获RTSP流并保存为多个5秒MP4文件？

首先，咱们先拆解你当前代码里的核心问题：

• 并发修改异常：bufferedImageList在两个线程中被同时读写（一个线程添加帧，另一个线程遍历并清空），LinkedList不是线程安全的，直接操作必然触发ConcurrentModificationException。
• 编码逻辑错误：每次遍历帧都调用addVideoStream，但这个方法应该是每个输出文件只调用一次的，重复调用会导致流ID冲突，也就是你遇到的Unable to create stream id 0错误。
• 全局Writer复用错误：你用了同一个IMediaWriter来处理所有片段，这会导致编码上下文混乱，每个5秒片段应该对应一个独立的Writer实例。
• Xuggler的维护问题：Xuggler已经多年没有更新了，社区支持和兼容性都很差，推荐改用更活跃的Java多媒体库，比如JavaCV（基于FFmpeg和OpenCV，稳定且功能完整）。

接下来，我会用JavaCV + RxJava来实现你的需求，这样既解决并发问题，又能优雅处理流的分段：

实现思路 #

1. 用JavaCV的FFmpegFrameGrabber直接拉取RTSP流，比通过Webcam+VLCJ的方式更高效、更直接。
2. 用RxJava的Flowable来处理帧流，通过buffer操作符按时间（5秒）或者帧数来分割片段。
3. 每个片段创建独立的FFmpegFrameRecorder来编码保存为MP4文件，避免流ID冲突。
4. 全程用RxJava的线程调度来处理并发，避免手动线程管理的混乱。

完整代码实现 #

首先需要添加JavaCV的依赖（Maven为例）：

<dependency>
    <groupId>org.bytedeco</groupId>
    <artifactId>javacv-platform</artifactId>
    <version>1.5.10</version>
</dependency>
<dependency>
    <groupId>io.reactivex.rxjava3</groupId>
    <artifactId>rxjava</artifactId>
    <version>3.1.8</version>
</dependency>

然后是核心代码：

import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.schedulers.Schedulers;
import org.bytedeco.ffmpeg.global.avcodec;
import org.bytedeco.javacv.FFmpegFrameGrabber;
import org.bytedeco.javacv.FFmpegFrameRecorder;
import org.bytedeco.javacv.Frame;

import java.io.File;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class RTSPStreamSplitter {
    private static final String RTSP_URL = "rtsp://wowzaec2demo.streamlock.net/vod/mp4:BigBuckBunny_175k.mov";
    private static final String OUTPUT_FOLDER = "C:\\Image\\";
    private static final int SEGMENT_DURATION_SECONDS = 5;
    private static final DateTimeFormatter FILE_NAME_FORMATTER = DateTimeFormatter.ofPattern("yyyyMMdd_HHmmss");

    public static void main(String[] args) throws Exception {
        // 创建输出文件夹
        File outputDir = new File(OUTPUT_FOLDER);
        if (!outputDir.exists()) {
            outputDir.mkdirs();
        }

        // 初始化RTSP抓取器
        FFmpegFrameGrabber grabber = new FFmpegFrameGrabber(RTSP_URL);
        grabber.start();

        // 获取流的基本信息
        int frameWidth = grabber.getImageWidth();
        int frameHeight = grabber.getImageHeight();
        double frameRate = grabber.getFrameRate();
        if (frameRate <= 0) {
            frameRate = 30.0;
        }

        // 用RxJava将帧流转换为Flowable，并按时间分段
        Flowable.create(emitter -> {
            Frame frame;
            try {
                while ((frame = grabber.grabFrame()) != null && !emitter.isCancelled()) {
                    // 只处理视频帧
                    if (frame.image != null) {
                        emitter.onNext(frame);
                    }
                }
                emitter.onComplete();
            } catch (Exception e) {
                emitter.onError(e);
            } finally {
                grabber.stop();
                grabber.release();
            }
        }, FlowableEmitter.BackpressureMode.BUFFER)
                // 按时间分组，每SEGMENT_DURATION_SECONDS秒一个片段
                .buffer(SEGMENT_DURATION_SECONDS, TimeUnit.SECONDS)
                .observeOn(Schedulers.io())
                .subscribe(
                        frameList -> saveSegment(frameList, frameWidth, frameHeight, frameRate),
                        error -> {
                            System.err.println("处理流时出错: " + error.getMessage());
                            error.printStackTrace();
                        },
                        () -> System.out.println("流处理完成")
                );

        // 保持主线程存活，直到手动终止
        synchronized (new Object()) {
            wait();
        }
    }

    private static void saveSegment(List<Frame> frameList, int width, int height, double frameRate) {
        if (frameList.isEmpty()) {
            return;
        }

        // 生成唯一的文件名，基于当前时间
        String fileName = OUTPUT_FOLDER + "segment_" + FILE_NAME_FORMATTER.format(LocalDateTime.now()) + ".mp4";
        FFmpegFrameRecorder recorder = new FFmpegFrameRecorder(fileName, width, height);

        try {
            // 配置编码器参数
            recorder.setVideoCodec(avcodec.AV_CODEC_ID_H264);
            recorder.setFormat("mp4");
            recorder.setFrameRate(frameRate);
            recorder.setVideoBitrate(175000); // 和原流匹配的码率
            recorder.start();

            // 写入当前片段的所有帧
            for (Frame frame : frameList) {
                recorder.record(frame);
            }

            System.out.println("已保存片段: " + fileName);
        } catch (Exception e) {
            System.err.println("保存片段失败: " + fileName);
            e.printStackTrace();
        } finally {
            try {
                recorder.stop();
                recorder.release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

代码说明 #

1. RTSP抓取：用FFmpegFrameGrabber直接连接RTSP流，跳过了Webcam中间层，性能和稳定性更好。
2. RxJava流处理：
   • Flowable.create将帧抓取转换为响应式流，自动处理背压。
   • buffer(5, TimeUnit.SECONDS)自动将每5秒的帧打包成一个列表，完美实现分段需求。
   • observeOn(Schedulers.io())让片段保存操作在IO线程执行，不阻塞主线程。
3. 编码保存：每个片段创建独立的FFmpegFrameRecorder，配置H264编码器（比MPEG4更高效），确保每个MP4文件的流上下文独立，避免编码冲突。
4. 线程安全：RxJava的调度器和背压机制自动处理并发问题，无需手动同步集合。

额外优化建议 #

• 如果需要更精确的分段（比如严格按帧数而不是时间），可以改用buffer((int)(frameRate * SEGMENT_DURATION_SECONDS))。
• 可以添加错误重试机制，比如当RTSP流断开时自动重连。
• 可以限制同时保存的片段数量，避免IO线程过载，比如用flatMap配合maxConcurrency参数。

内容的提问来源于stack exchange，提问作者quarks