diff --git a/include/data_managet.hpp b/include/data_managet.hpp
new file mode 100644
index 0000000..f553b23
--- /dev/null
+++ b/include/data_managet.hpp
@@ -0,0 +1,18 @@
+#pragma once
+#include <string>
+#include <vector>
+
+struct VideoPushRequest
+{
+    std::string type;
+    std::string seqNo;
+
+    struct DataItem
+    {
+        int switchVal;  // 0=开始推流, 1=停止推流
+        std::vector<int> channels;
+        int streamType;  // 0=主, 1=子
+    };
+
+    std::vector<DataItem> data;
+};
diff --git a/include/mqtt_client_wrapper.hpp b/include/mqtt_client_wrapper.hpp
index 9a566cd..4daaf7b 100644
--- a/include/mqtt_client_wrapper.hpp
+++ b/include/mqtt_client_wrapper.hpp
@@ -1,29 +1,15 @@
-// mqtt_client_wrapper.hppa
+// mqtt_client_wrapper.hpp
 #pragma once
 
 #include <atomic>
 #include <memory>
 
 #include "app_config.hpp"
+#include "data_managet.hpp"
 #include "logger.hpp"
 #include "mqtt_client.hpp"
 #include "rtmp_manager.hpp"
 
-struct VideoPushRequest
-{
-    std::string type;   // "request"
-    std::string seqNo;  // 流水号
-
-    struct DataItem
-    {
-        int switchVal;              // 0=开始推流, 1=停止推流
-        std::vector<int> channels;  // 需要推送的视频通道
-        int streamType;             // 0=主码流, 1=子码流
-    };
-
-    std::vector<DataItem> data;  // 可以支持多条任务（即一条消息既开又关多通道）
-};
-
 // 启动 MQTT 客户端线程（内部自动重连、订阅等）
 void mqtt_client_thread_func();
 
diff --git a/include/rtmp_manager.hpp b/include/rtmp_manager.hpp
index c6d57e2..242f1e8 100644
--- a/include/rtmp_manager.hpp
+++ b/include/rtmp_manager.hpp
@@ -1,54 +1,89 @@
 // rtmp_manager.hpp
 #pragma once
 
+#include <gst/gst.h>
+
 #include <atomic>
-#include <condition_variable>
-#include <future>
+#include <functional>
 #include <memory>
 #include <mutex>
-#include <queue>
 #include <string>
 #include <thread>
 #include <unordered_map>
 #include <vector>
 
 #include "app_config.hpp"
-#include "mqtt_client_wrapper.hpp"
+#include "data_manager.hpp"
 
 class RTMPManager
 {
    public:
+    enum class StreamResult
+    {
+        OK,
+        PIPELINE_ERROR,
+        CONNECTION_FAIL,
+        EOS_RECEIVED,
+        TIMEOUT,
+        UNKNOWN
+    };
+
+    struct StreamStatus
+    {
+        bool running{false};
+        StreamResult last_result{StreamResult::UNKNOWN};
+        std::string last_error;
+    };
+
     struct StreamResultInfo
     {
         int loc{-1};
         std::string url;
-        int result{1};  // 0=成功, 1=失败
+        int result{1};  // 0 success, 1 fail
         std::string reason;
     };
 
+    using StreamCallback = std::function<void(const std::string &key, StreamStatus)>;
+
     static void init();
-    static void enqueue_video_push_request(const VideoPushRequest &req);
+    static void set_status_callback(StreamCallback cb);
+
+    // 单路接口
+    static StreamResultInfo start_camera(const Camera &cam, StreamType type);
+    static StreamResultInfo stop_camera(const std::string &cam_name, StreamType type);
+
+    // 批量接口：处理 VideoPushRequest
+    static std::vector<StreamResultInfo> process_push_request(const VideoPushRequest &req);
+
     static void stop_all();
+    static bool is_streaming(const std::string &cam_name, StreamType type);
+    static bool is_any_streaming();
+    static std::string get_stream_url(const std::string &cam_name, StreamType type);
 
    private:
     struct StreamContext
     {
         std::atomic<bool> running{false};
         std::thread thread;
-        std::promise<StreamResultInfo> start_result;
+        StreamStatus status;
+        int retry_count{0};
+
+        StreamContext() = default;
+        StreamContext(const StreamContext &) = delete;
+        StreamContext &operator=(const StreamContext &) = delete;
     };
 
-    static void rtmp_worker_thread();
-    static StreamResultInfo start_camera(const Camera &cam, int streamType);
-    static StreamResultInfo stop_camera(const Camera &cam, int streamType);
+    static std::string make_stream_key(const std::string &cam_name, StreamType type);
+    static GstElement *create_pipeline(const Camera &cam, StreamType type);
+    static void stream_loop(Camera cam, StreamType type);
+
+    static void update_status(const std::string &key, const StreamStatus &status);
 
     static std::unordered_map<std::string, std::unique_ptr<StreamContext>> streams;
     static std::mutex streams_mutex;
+    static StreamCallback status_callback;
 
-    static std::queue<VideoPushRequest> request_queue;
-    static std::mutex queue_mutex;
-    static std::condition_variable queue_cv;
-
-    static std::thread worker_thread;
-    static std::atomic<bool> running;
+    // 自动重试参数
+    static constexpr int MAX_RETRIES = 3;
+    static constexpr int RETRY_DELAY_MS = 2000;
 };
\ No newline at end of file
diff --git a/src/mqtt_client_wrapper.cpp b/src/mqtt_client_wrapper.cpp
index 6cc8a70..2d8d03e 100644
--- a/src/mqtt_client_wrapper.cpp
+++ b/src/mqtt_client_wrapper.cpp
@@ -83,7 +83,33 @@ static void on_mqtt_message_received(const std::string &topic, const std::string
             }
         }
 
-        RTMPManager::enqueue_video_push_request(req);
+        std::thread(
+            [req]()
+            {
+                // 调用 RTMP 模块执行批量推流启停
+                auto results = RTMPManager::process_push_request(req);
+
+                // 组装 MQTT 回复
+                nlohmann::json reply;
+                reply["type"] = "response";
+                reply["seqNo"] = req.seqNo;
+                reply["data"] = nlohmann::json::array();
+
+                for (const auto &r : results)
+                {
+                    nlohmann::json item;
+                    item["loc"] = r.loc;
+                    item["url"] = r.url;
+                    item["result"] = r.result;
+                    item["reason"] = r.reason;
+                    reply["data"].push_back(item);
+                }
+
+                // 发送回复到 MQTT
+                mqtt_client.publish(g_app_config.mqtt.topics.video_up, reply.dump());
+                LOG_INFO("[MQTT] Sent RTMP response: " + reply.dump());
+            })
+            .detach();
     }
     catch (const std::exception &e)
     {
diff --git a/src/rtmp_manager.cpp b/src/rtmp_manager.cpp
index 8107e94..0c9c730 100644
--- a/src/rtmp_manager.cpp
+++ b/src/rtmp_manager.cpp
@@ -1,105 +1,94 @@
-// rtmp_manager.cpp
+// rtsp_manager.cpp
 #include "rtmp_manager.hpp"
 
-#include <gst/gst.h>
+#include <chrono>
+#include <iostream>
+#include <thread>
 
 #include "logger.hpp"
 
+static inline std::string stream_type_suffix(StreamType type) { return (type == StreamType::MAIN) ? "_main" : "_sub"; }
+
 std::unordered_map<std::string, std::unique_ptr<RTMPManager::StreamContext>> RTMPManager::streams;
 std::mutex RTMPManager::streams_mutex;
+RTMPManager::StreamCallback RTMPManager::status_callback = nullptr;
 
-std::queue<VideoPushRequest> RTMPManager::request_queue;
-std::mutex RTMPManager::queue_mutex;
-std::condition_variable RTMPManager::queue_cv;
-
-std::thread RTMPManager::worker_thread;
-std::atomic<bool> RTMPManager::running{false};
+std::string RTMPManager::make_stream_key(const std::string &cam_name, StreamType type)
+{
+    return cam_name + stream_type_suffix(type);
+}
 
 void RTMPManager::init()
 {
     gst_init(nullptr, nullptr);
-    running = true;
-    worker_thread = std::thread(rtmp_worker_thread);
-    LOG_INFO("[RTMP] Initialized and worker thread started.");
+    LOG_INFO("[RTMP] GStreamer initialized.");
 }
 
-void RTMPManager::stop_all()
-{
-    running = false;
-    queue_cv.notify_all();
+void RTMPManager::set_status_callback(StreamCallback cb) { status_callback = std::move(cb); }
 
-    // 停止所有流
+void RTMPManager::update_status(const std::string &key, const StreamStatus &status)
+{
     {
         std::lock_guard<std::mutex> lock(streams_mutex);
-        for (auto &kv : streams) kv.second->running = false;
+        auto it = streams.find(key);
+        if (it != streams.end()) it->second->status = status;
     }
-
-    if (worker_thread.joinable()) worker_thread.join();
-
-    LOG_INFO("[RTMP] stop_all completed.");
+    if (status_callback) status_callback(key, status);
 }
 
-void RTMPManager::enqueue_video_push_request(const VideoPushRequest &req)
+GstElement *RTMPManager::create_pipeline(const Camera &cam, StreamType type)
 {
+    int width = cam.width, height = cam.height, fps = cam.fps, bitrate = cam.bitrate;
+    if (type == StreamType::SUB)
     {
-        std::lock_guard<std::mutex> lock(queue_mutex);
-        request_queue.push(req);
+        width = std::max(160, width / 2);
+        height = std::max(120, height / 2);
+        fps = std::max(10, fps / 2);
+        bitrate = std::max(300000, bitrate / 2);
     }
-    queue_cv.notify_one();
+
+    std::string stream_name = cam.name + stream_type_suffix(type);
+    std::string pipeline_str = "v4l2src device=" + cam.device +
+                               " ! video/x-raw,format=NV12,width=" + std::to_string(width) +
+                               ",height=" + std::to_string(height) + ",framerate=" + std::to_string(fps) +
+                               "/1 ! queue max-size-buffers=1 leaky=downstream "
+                               "! mpph264enc bps=" +
+                               std::to_string(bitrate) + " gop=" + std::to_string(fps) +
+                               " ! h264parse ! flvmux streamable=true name=mux "
+                               "! rtmpsink location=\"rtmp://127.0.0.1/live/" +
+                               stream_name + " live=1\" sync=false";
+
+    LOG_INFO("[RTMP] Creating pipeline for '" + stream_name + "': " + pipeline_str);
+
+    GError *error = nullptr;
+    GstElement *pipeline = gst_parse_launch(pipeline_str.c_str(), &error);
+    if (error)
+    {
+        LOG_ERROR("[RTMP] Pipeline parse error: " + std::string(error->message));
+        g_error_free(error);
+        return nullptr;
+    }
+    return pipeline;
 }
 
-void RTMPManager::rtmp_worker_thread()
+int get_camera_index(const std::string &name)
 {
-    while (running)
-    {
-        VideoPushRequest req;
-        {
-            std::unique_lock<std::mutex> lock(queue_mutex);
-            queue_cv.wait(lock, [] { return !request_queue.empty() || !running; });
-            if (!running) break;
-            req = request_queue.front();
-            request_queue.pop();
-        }
-
-        nlohmann::json results = nlohmann::json::array();
-
-        for (auto &item : req.data)
-        {
-            for (int ch : item.channels)
-            {
-                if (ch < 0 || ch >= static_cast<int>(g_app_config.cameras.size())) continue;
-                Camera &cam = g_app_config.cameras[ch];
-                StreamResultInfo res;
-
-                if (item.switchVal == 0)
-                    res = start_camera(cam, item.streamType);
-                else
-                    res = stop_camera(cam, item.streamType);
-
-                res.loc = ch;
-                res.url = "rtmp://127.0.0.1/live/" + cam.name;
-                results.push_back({{"loc", res.loc}, {"url", res.url}, {"result", res.result}, {"reason", res.reason}});
-            }
-        }
-
-        // 上报 MQTT
-        nlohmann::json reply;
-        reply["type"] = "response";
-        reply["seqNo"] = req.seqNo;
-        reply["data"] = results;
-        if (mqtt_client) mqtt_client->publish(g_app_config.mqtt.topics.video_down, reply.dump());
-    }
+    for (size_t i = 0; i < g_app_config.cameras.size(); ++i)
+        if (g_app_config.cameras[i].name == name) return static_cast<int>(i);
+    return -1;
 }
 
-// 简化 start_camera / stop_camera，不阻塞 MQTT 回调
-RTMPManager::StreamResultInfo RTMPManager::start_camera(const Camera &cam, int streamType)
+RTMPManager::StreamResultInfo RTMPManager::start_camera(const Camera &cam, StreamType type)
 {
     StreamResultInfo res;
-    std::string key = cam.name + (streamType == 0 ? "_main" : "_sub");
+    res.loc = get_camera_index(cam.name);
+    res.url = get_stream_url(cam.name, type);
 
+    std::string key = make_stream_key(cam.name, type);
     {
         std::lock_guard<std::mutex> lock(streams_mutex);
-        if (streams.find(key) != streams.end())
+        auto it = streams.find(key);
+        if (it != streams.end() && it->second->running.load())
         {
             res.result = 0;
             res.reason = "Already streaming";
@@ -107,41 +96,24 @@ RTMPManager::StreamResultInfo RTMPManager::start_camera(const Camera &cam, int s
         }
 
         auto ctx = std::make_unique<StreamContext>();
-        ctx->running = true;
-        ctx->thread = std::thread(
-            [cam, streamType, ctx_ptr = ctx.get()]
-            {
-                // TODO: GStreamer pipeline loop
-                std::this_thread::sleep_for(std::chrono::seconds(1));  // 模拟启动
-                StreamResultInfo r;
-                r.result = 0;
-                r.reason = "Started OK";
-                ctx_ptr->start_result.set_value(r);
-            });
-
-        streams[key] = std::move(ctx);
-    }
-
-    // 非阻塞：直接返回 future 的结果即可
-    try
-    {
-        res = streams[key]->start_result.get_future().get();
-    }
-    catch (...)
-    {
-        res.result = 1;
-        res.reason = "Exception in start_camera";
+        ctx->running.store(true);
+        ctx->thread = std::thread([cam, type]() { stream_loop(cam, type); });
+        streams.emplace(key, std::move(ctx));
     }
 
+    res.result = 0;
+    res.reason = "Started OK";
     return res;
 }
 
-RTMPManager::StreamResultInfo RTMPManager::stop_camera(const Camera &cam, int streamType)
+RTMPManager::StreamResultInfo RTMPManager::stop_camera(const std::string &cam_name, StreamType type)
 {
     StreamResultInfo res;
-    std::string key = cam.name + (streamType == 0 ? "_main" : "_sub");
+    res.loc = get_camera_index(cam_name);
+    res.url = get_stream_url(cam_name, type);
 
     std::unique_ptr<StreamContext> ctx;
+    std::string key = make_stream_key(cam_name, type);
     {
         std::lock_guard<std::mutex> lock(streams_mutex);
         auto it = streams.find(key);
@@ -152,13 +124,201 @@ RTMPManager::StreamResultInfo RTMPManager::stop_camera(const Camera &cam, int st
             return res;
         }
 
+        it->second->running.store(false);
         ctx = std::move(it->second);
         streams.erase(it);
-        ctx->running = false;
     }
 
     if (ctx->thread.joinable()) ctx->thread.join();
+
     res.result = 0;
-    res.reason = "Stopped OK";
+    res.reason = "Stopped manually";
     return res;
-}
\ No newline at end of file
+}
+
+void RTMPManager::stop_all()
+{
+    std::vector<std::unique_ptr<StreamContext>> ctxs;
+    {
+        std::lock_guard<std::mutex> lock(streams_mutex);
+        for (auto &kv : streams) kv.second->running.store(false);
+        for (auto &kv : streams) ctxs.push_back(std::move(kv.second));
+        streams.clear();
+    }
+
+    for (auto &ctx : ctxs)
+        if (ctx->thread.joinable()) ctx->thread.join();
+
+    LOG_INFO("[RTMP] stop_all completed.");
+}
+
+bool RTMPManager::is_streaming(const std::string &cam_name, StreamType type)
+{
+    std::lock_guard<std::mutex> lock(streams_mutex);
+    auto it = streams.find(make_stream_key(cam_name, type));
+    return it != streams.end() && it->second->running.load();
+}
+
+bool RTMPManager::is_any_streaming()
+{
+    std::lock_guard<std::mutex> lock(streams_mutex);
+    for (auto &kv : streams)
+        if (kv.second->running.load()) return true;
+    return false;
+}
+
+std::string RTMPManager::get_stream_url(const std::string &cam_name, StreamType type)
+{
+    return "rtmp://127.0.0.1/live/" + make_stream_key(cam_name, type);
+}
+
+void RTMPManager::stream_loop(Camera cam, StreamType type)
+{
+    std::string key = make_stream_key(cam.name, type);
+    LOG_INFO("[RTMP] Stream loop started for " + key);
+
+    int retries = 0;
+
+    while (true)
+    {
+        {
+            std::lock_guard<std::mutex> lock(streams_mutex);
+            auto it = streams.find(key);
+            if (it == streams.end() || !it->second->running.load()) break;
+        }
+
+        GstElement *pipeline = create_pipeline(cam, type);
+        if (!pipeline)
+        {
+            update_status(key, {false, StreamResult::PIPELINE_ERROR, "Failed to create pipeline"});
+            if (++retries <= MAX_RETRIES)
+            {
+                LOG_WARN("[RTMP] Retrying pipeline creation for " + key);
+                std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_DELAY_MS));
+                continue;
+            }
+            break;
+        }
+
+        GstBus *bus = gst_element_get_bus(pipeline);
+        gst_element_set_state(pipeline, GST_STATE_PLAYING);
+
+        bool first_frame_received = false;
+        bool stop_flag = false;
+        auto first_frame_start = std::chrono::steady_clock::now();
+
+        while (!stop_flag)
+        {
+            GstMessage *msg = gst_bus_timed_pop_filtered(
+                bus, 100 * GST_MSECOND,
+                static_cast<GstMessageType>(GST_MESSAGE_ERROR | GST_MESSAGE_EOS | GST_MESSAGE_STATE_CHANGED));
+
+            {
+                std::lock_guard<std::mutex> lock(streams_mutex);
+                auto it = streams.find(key);
+                if (it == streams.end() || !it->second->running.load())
+                {
+                    stop_flag = true;
+                    break;
+                }
+            }
+
+            if (!first_frame_received)
+            {
+                auto elapsed = std::chrono::steady_clock::now() - first_frame_start;
+                if (std::chrono::duration_cast<std::chrono::seconds>(elapsed).count() > 5)
+                {
+                    update_status(key, {false, StreamResult::TIMEOUT, "No frames received"});
+                    stop_flag = true;
+                }
+            }
+
+            if (!msg) continue;
+
+            switch (GST_MESSAGE_TYPE(msg))
+            {
+                case GST_MESSAGE_ERROR:
+                {
+                    GError *err = nullptr;
+                    gst_message_parse_error(msg, &err, nullptr);
+                    update_status(key, {false, StreamResult::CONNECTION_FAIL, err ? err->message : "Unknown error"});
+                    if (err) g_error_free(err);
+                    stop_flag = true;
+                    break;
+                }
+                case GST_MESSAGE_EOS:
+                    update_status(key, {false, StreamResult::EOS_RECEIVED, "EOS"});
+                    stop_flag = true;
+                    break;
+                case GST_MESSAGE_STATE_CHANGED:
+                {
+                    GstState old_state, new_state;
+                    gst_message_parse_state_changed(msg, &old_state, &new_state, nullptr);
+                    if (GST_MESSAGE_SRC(msg) == GST_OBJECT(pipeline) && new_state == GST_STATE_PLAYING &&
+                        !first_frame_received)
+                    {
+                        first_frame_received = true;
+                        update_status(key, {true, StreamResult::OK, ""});
+                        retries = 0;  // 成功后重置重试计数
+                    }
+                    break;
+                }
+                default:
+                    break;
+            }
+
+            gst_message_unref(msg);
+        }
+
+        gst_element_set_state(pipeline, GST_STATE_NULL);
+        if (bus) gst_object_unref(bus);
+        gst_object_unref(pipeline);
+
+        // 自动重试逻辑
+        if (stop_flag && retries < MAX_RETRIES)
+        {
+            retries++;
+            LOG_WARN("[RTMP] Stream " + key + " failed, retrying... (" + std::to_string(retries) + ")");
+            std::this_thread::sleep_for(std::chrono::milliseconds(RETRY_DELAY_MS));
+            continue;
+        }
+
+        break;
+    }
+
+    update_status(key, {false, StreamResult::UNKNOWN, "Stream loop exited"});
+    LOG_INFO("[RTMP] Stream loop ended for " + key);
+}
+
+std::vector<RTMPManager::StreamResultInfo> RTMPManager::process_push_request(const VideoPushRequest &req)
+{
+    std::vector<StreamResultInfo> results;
+
+    for (const auto &item : req.data)
+    {
+        StreamType type = (item.streamType == 0) ? StreamType::MAIN : StreamType::SUB;
+
+        for (int ch : item.channels)
+        {
+            if (ch < 0 || ch >= static_cast<int>(g_app_config.cameras.size())) continue;
+
+            const auto &cam = g_app_config.cameras[ch];
+            StreamResultInfo info;
+
+            if (item.switchVal == 0)
+            {
+                // 开启推流
+                info = start_camera(cam, type);
+            }
+            else
+            {
+                // 停止推流
+                info = stop_camera(cam.name, type);
+            }
+
+            results.push_back(info);
+        }
+    }
+
+    return results;
+}