diff --git a/src/Common/MediaSource.cpp b/src/Common/MediaSource.cpp
index 4a3b4373..3437de93 100644
--- a/src/Common/MediaSource.cpp
+++ b/src/Common/MediaSource.cpp
@@ -233,10 +233,17 @@ toolkit::EventPoller::Ptr MediaSource::getOwnerPoller() {
 }
 
 void MediaSource::onReaderChanged(int size) {
-    auto listener = _listener.lock();
-    if (listener) {
-        listener->onReaderChanged(*this, size);
-    }
+    weak_ptr<MediaSource> weak_self = shared_from_this();
+    getOwnerPoller()->async([weak_self, size]() {
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            return;
+        }
+        auto listener = strong_self->_listener.lock();
+        if (listener) {
+            listener->onReaderChanged(*strong_self, size);
+        }
+    });
 }
 
 bool MediaSource::setupRecord(Recorder::type type, bool start, const string &custom_path, size_t max_second){
diff --git a/src/Common/MultiMediaSourceMuxer.cpp b/src/Common/MultiMediaSourceMuxer.cpp
index 88886330..cdfdde39 100644
--- a/src/Common/MultiMediaSourceMuxer.cpp
+++ b/src/Common/MultiMediaSourceMuxer.cpp
@@ -230,7 +230,6 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &, const MediaSourceEvent::
             rtp_sender->addTrack(track);
         }
         rtp_sender->addTrackCompleted();
-        lock_guard<mutex> lck(strong_self->_rtp_sender_mtx);
         strong_self->_rtp_sender[args.ssrc] = rtp_sender;
     });
 #else
@@ -249,13 +248,11 @@ bool MultiMediaSourceMuxer::stopSendRtp(MediaSource &sender, const string &ssrc)
     }
     if (ssrc.empty()) {
         //关闭全部
-        lock_guard<mutex> lck(_rtp_sender_mtx);
         auto size = _rtp_sender.size();
         _rtp_sender.clear();
         return size;
     }
     //关闭特定的
-    lock_guard<mutex> lck(_rtp_sender_mtx);
     return _rtp_sender.erase(ssrc);
 #else
     return false;
@@ -340,7 +337,6 @@ void MultiMediaSourceMuxer::resetTracks() {
 #endif
 
 #if defined(ENABLE_RTPPROXY)
-    lock_guard<mutex> lck(_rtp_sender_mtx);
     for (auto &pr : _rtp_sender) {
         pr.second->resetTracks();
     }
@@ -447,7 +443,6 @@ bool MultiMediaSourceMuxer::onTrackFrame(const Frame::Ptr &frame_in) {
 #endif
 
 #if defined(ENABLE_RTPPROXY)
-    lock_guard<mutex> lck(_rtp_sender_mtx);
     for (auto &pr : _rtp_sender) {
         ret = pr.second->inputFrame(frame) ? true : ret;
     }
diff --git a/src/Common/MultiMediaSourceMuxer.h b/src/Common/MultiMediaSourceMuxer.h
index 92c98f1b..caedd25d 100644
--- a/src/Common/MultiMediaSourceMuxer.h
+++ b/src/Common/MultiMediaSourceMuxer.h
@@ -180,7 +180,6 @@ private:
     std::weak_ptr<Listener> _track_listener;
     std::function<std::string()> _get_origin_url;
 #if defined(ENABLE_RTPPROXY)
-    std::mutex _rtp_sender_mtx;
     std::unordered_map<std::string, RtpSender::Ptr> _rtp_sender;
 #endif //ENABLE_RTPPROXY
 
diff --git a/src/Rtp/RtpProcess.cpp b/src/Rtp/RtpProcess.cpp
index 511c9581..36a796b8 100644
--- a/src/Rtp/RtpProcess.cpp
+++ b/src/Rtp/RtpProcess.cpp
@@ -277,5 +277,9 @@ std::shared_ptr<SockInfo> RtpProcess::getOriginSock(MediaSource &sender) const {
     return const_cast<RtpProcess *>(this)->shared_from_this();
 }
 
+toolkit::EventPoller::Ptr RtpProcess::getOwnerPoller(MediaSource &sender) {
+    return _sock ? _sock->getPoller() : nullptr;
+}
+
 }//namespace mediakit
 #endif//defined(ENABLE_RTPPROXY)
\ No newline at end of file
diff --git a/src/Rtp/RtpProcess.h b/src/Rtp/RtpProcess.h
index c9a992b9..93dd37fe 100644
--- a/src/Rtp/RtpProcess.h
+++ b/src/Rtp/RtpProcess.h
@@ -76,6 +76,7 @@ protected:
     MediaOriginType getOriginType(MediaSource &sender) const override;
     std::string getOriginUrl(MediaSource &sender) const override;
     std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
+    toolkit::EventPoller::Ptr getOwnerPoller(MediaSource &sender) override;
 
 private:
     void emitOnPublish();