添加合并写功能

src/Common/MediaSource.cpp

@@ -450,4 +450,50 @@ MediaSource::Ptr MediaSource::createFromMP4(const string &schema, const string &
 #endif //ENABLE_MP4
 }
 
+static bool isFlushAble_default(bool is_audio, uint32_t last_stamp, uint32_t new_stamp, int cache_size) {
+    if (new_stamp < last_stamp) {
+        //时间戳回退(可能seek中)
+        return true;
+    }
+
+    if (!is_audio) {
+        //这是视频,时间戳发送变化或者缓存超过1024个
+        return last_stamp != new_stamp || cache_size >= 1024;
+    }
+
+    //这是音频,缓存超过100ms或者缓存个数超过10个
+    return new_stamp > last_stamp + 100 || cache_size > 10;
+}
+
+static bool isFlushAble_merge(bool is_audio, uint32_t last_stamp, uint32_t new_stamp, int cache_size, int merge_ms) {
+    if (new_stamp < last_stamp) {
+        //时间戳回退(可能seek中)
+        return true;
+    }
+
+    if(new_stamp > last_stamp + merge_ms){
+        //时间戳增量超过合并写阈值
+        return true;
+    }
+
+    if (!is_audio) {
+        //这是视频,缓存数超过1024个,这个逻辑用于避免时间戳异常的流导致的内存暴增问题
+        //而且sendmsg接口一般最多只能发送1024个数据包
+        return cache_size >= 1024;
+    }
+
+    //这是音频，音频缓存超过20个
+    return cache_size > 20;
+}
+
+bool FlushPolicy::isFlushAble(uint32_t last_stamp, uint32_t new_stamp, int cache_size) {
+    GET_CONFIG(bool,ultraLowDelay, General::kUltraLowDelay);
+    GET_CONFIG(int,mergeWriteMS, General::kMergeWriteMS);
+    if(ultraLowDelay || mergeWriteMS <= 0){
+        //关闭了合并写或者合并写阈值小于等于0
+        return isFlushAble_default(_is_audio, last_stamp, new_stamp, cache_size);
+    }
+    return isFlushAble_merge(_is_audio, last_stamp, new_stamp, cache_size,mergeWriteMS);
+}
+
 } /* namespace mediakit */

src/Common/MediaSource.h

@@ -21,6 +21,9 @@
 #include "Util/logger.h"
 #include "Util/TimeTicker.h"
 #include "Util/NoticeCenter.h"
+#include "Util/List.h"
+#include "Rtsp/Rtsp.h"
+#include "Rtmp/Rtmp.h"
 #include "Extension/Track.h"
 #include "Record/Recorder.h"
 
@@ -153,6 +156,114 @@ private:
     static recursive_mutex g_mtxMediaSrc;
 };
 
+///缓存刷新策略类
+class FlushPolicy {
+public:
+    FlushPolicy(bool is_audio) {
+        _is_audio = is_audio;
+    };
+
+    ~FlushPolicy() = default;
+
+    uint32_t getStamp(const RtpPacket::Ptr &packet) {
+        return packet->timeStamp;
+    }
+
+    uint32_t getStamp(const RtmpPacket::Ptr &packet) {
+        return packet->timeStamp;
+    }
+
+    bool isFlushAble(uint32_t last_stamp, uint32_t new_stamp, int cache_size);
+private:
+    bool _is_audio;
+};
+
+/// 视频合并写缓存模板
+/// \tparam packet 包类型
+/// \tparam policy 刷新缓存策略
+/// \tparam packet_list 包缓存类型
+template<typename packet, typename policy = FlushPolicy, typename packet_list = List<std::shared_ptr<packet> > >
+class VideoPacketCache {
+public:
+    VideoPacketCache() : _policy(true) {
+        _cache = std::make_shared<packet_list>();
+    }
+
+    virtual ~VideoPacketCache() = default;
+
+    void inputVideo(const std::shared_ptr<packet> &rtp, bool key_pos) {
+        auto new_stamp = _policy.getStamp(rtp);
+        if (_policy.isFlushAble(_last_stamp, new_stamp, _cache->size())) {
+            flushAll();
+        }
+
+        //追加数据到最后
+        _cache->emplace_back(rtp);
+        _last_stamp = new_stamp;
+        if (key_pos) {
+            _key_pos = key_pos;
+        }
+    }
+
+    virtual void onFlushVideo(std::shared_ptr<packet_list> &, bool key_pos) = 0;
+
+private:
+    void flushAll() {
+        if (_cache->empty()) {
+            return;
+        }
+        onFlushVideo(_cache, _key_pos);
+        _cache = std::make_shared<packet_list>();
+        _key_pos = false;
+    }
+
+private:
+    policy _policy;
+    std::shared_ptr<packet_list> _cache;
+    uint32_t _last_stamp = 0;
+    bool _key_pos = false;
+};
+
+/// 音频频合并写缓存模板
+/// \tparam packet 包类型
+/// \tparam policy 刷新缓存策略
+/// \tparam packet_list 包缓存类型
+template<typename packet, typename policy = FlushPolicy, typename packet_list = List<std::shared_ptr<packet> > >
+class AudioPacketCache {
+public:
+    AudioPacketCache() : _policy(false) {
+        _cache = std::make_shared<packet_list>();
+    }
+
+    virtual ~AudioPacketCache() = default;
+
+    void inputAudio(const std::shared_ptr<packet> &rtp) {
+        auto new_stamp = _policy.getStamp(rtp);
+        if (_policy.isFlushAble(_last_stamp, new_stamp, _cache->size())) {
+            flushAll();
+        }
+        //追加数据到最后
+        _cache->emplace_back(rtp);
+        _last_stamp = new_stamp;
+    }
+
+    virtual void onFlushAudio(std::shared_ptr<packet_list> &) = 0;
+
+private:
+    void flushAll() {
+        if (_cache->empty()) {
+            return;
+        }
+        onFlushAudio(_cache);
+        _cache = std::make_shared<packet_list>();
+    }
+
+private:
+    policy _policy;
+    std::shared_ptr<packet_list> _cache;
+    uint32_t _last_stamp = 0;
+};
+
 } /* namespace mediakit */
 
 

src/Common/config.cpp

@@ -67,6 +67,7 @@ const string kResetWhenRePlay = GENERAL_FIELD"resetWhenRePlay";
 const string kPublishToRtxp = GENERAL_FIELD"publishToRtxp";
 const string kPublishToHls = GENERAL_FIELD"publishToHls";
 const string kPublishToMP4 = GENERAL_FIELD"publishToMP4";
+const string kMergeWriteMS = GENERAL_FIELD"mergeWriteMS";
 
 onceToken token([](){
     mINI::Instance()[kFlowThreshold] = 1024;
@@ -79,6 +80,7 @@ onceToken token([](){
     mINI::Instance()[kPublishToRtxp] = 1;
     mINI::Instance()[kPublishToHls] = 1;
     mINI::Instance()[kPublishToMP4] = 0;
+    mINI::Instance()[kMergeWriteMS] = 300;
 },nullptr);
 
 }//namespace General

src/Common/config.h

@@ -173,6 +173,9 @@ extern const string kPublishToRtxp ;
 extern const string kPublishToHls ;
 //是否默认推流时mp4录像，hook接口(on_publish)中可以覆盖该设置
 extern const string kPublishToMP4 ;
+//合并写缓存大小(单位毫秒)，合并写指服务器缓存一定的数据后才会一次性写入socket，这样能提高性能，但是会提高延时
+//在开启低延时模式后，该参数不起作用
+extern const string kMergeWriteMS ;
 }//namespace General