重构SteamMessageHandler,移除多余线程,改用定时器实现异步轮询,优化消息接收逻辑

This commit is contained in:
Ayndpa
2025-11-19 21:42:51 +08:00
parent 04ce1a4a03
commit e8a326be53
2 changed files with 35 additions and 33 deletions

View File

@@ -2,12 +2,11 @@
#include <iostream> #include <iostream>
#include <cstring> #include <cstring>
#include <chrono> #include <chrono>
#include <thread>
#include <steam_api.h> #include <steam_api.h>
#include <isteamnetworkingsockets.h> #include <isteamnetworkingsockets.h>
SteamMessageHandler::SteamMessageHandler(boost::asio::io_context& io_context, ISteamNetworkingSockets* interface, std::vector<HSteamNetConnection>& connections, std::mutex& connectionsMutex, bool& g_isHost, int& localPort) SteamMessageHandler::SteamMessageHandler(boost::asio::io_context& io_context, ISteamNetworkingSockets* interface, std::vector<HSteamNetConnection>& connections, std::mutex& connectionsMutex, bool& g_isHost, int& localPort)
: io_context_(io_context), m_pInterface_(interface), connections_(connections), connectionsMutex_(connectionsMutex), g_isHost_(g_isHost), localPort_(localPort), running_(false) {} : io_context_(io_context), m_pInterface_(interface), connections_(connections), connectionsMutex_(connectionsMutex), g_isHost_(g_isHost), localPort_(localPort), running_(false), currentPollInterval_(0) {}
SteamMessageHandler::~SteamMessageHandler() { SteamMessageHandler::~SteamMessageHandler() {
stop(); stop();
@@ -16,23 +15,15 @@ SteamMessageHandler::~SteamMessageHandler() {
void SteamMessageHandler::start() { void SteamMessageHandler::start() {
if (running_) return; if (running_) return;
running_ = true; running_ = true;
thread_ = std::thread([this]() { run(); }); timer_ = std::make_unique<boost::asio::steady_timer>(io_context_);
// Start io_context in a separate thread startAsyncPoll();
io_thread_ = std::thread([this]() {
auto work = boost::asio::make_work_guard(io_context_);
io_context_.run();
});
} }
void SteamMessageHandler::stop() { void SteamMessageHandler::stop() {
if (!running_) return; if (!running_) return;
running_ = false; running_ = false;
io_context_.stop(); if (timer_) {
if (thread_.joinable()) { timer_->cancel();
thread_.join();
}
if (io_thread_.joinable()) {
io_thread_.join();
} }
} }
@@ -43,20 +34,14 @@ std::shared_ptr<MultiplexManager> SteamMessageHandler::getMultiplexManager(HStea
return multiplexManagers_[conn]; return multiplexManagers_[conn];
} }
void SteamMessageHandler::run() { void SteamMessageHandler::startAsyncPoll() {
while (running_) { if (!running_) return;
// Poll networking
// Poll networking callbacks
m_pInterface_->RunCallbacks(); m_pInterface_->RunCallbacks();
// Receive messages // Receive messages and check if any were received
pollMessages(); int totalMessages = 0;
// Sleep a bit to avoid busy loop
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
}
void SteamMessageHandler::pollMessages() {
std::vector<HSteamNetConnection> currentConnections; std::vector<HSteamNetConnection> currentConnections;
{ {
std::lock_guard<std::mutex> lockConn(connectionsMutex_); std::lock_guard<std::mutex> lockConn(connectionsMutex_);
@@ -65,6 +50,7 @@ void SteamMessageHandler::pollMessages() {
for (auto conn : currentConnections) { for (auto conn : currentConnections) {
ISteamNetworkingMessage* pIncomingMsgs[10]; ISteamNetworkingMessage* pIncomingMsgs[10];
int numMsgs = m_pInterface_->ReceiveMessagesOnConnection(conn, pIncomingMsgs, 10); int numMsgs = m_pInterface_->ReceiveMessagesOnConnection(conn, pIncomingMsgs, 10);
totalMessages += numMsgs;
for (int i = 0; i < numMsgs; ++i) { for (int i = 0; i < numMsgs; ++i) {
ISteamNetworkingMessage* pIncomingMsg = pIncomingMsgs[i]; ISteamNetworkingMessage* pIncomingMsg = pIncomingMsgs[i];
const char* data = (const char*)pIncomingMsg->m_pData; const char* data = (const char*)pIncomingMsg->m_pData;
@@ -77,4 +63,21 @@ void SteamMessageHandler::pollMessages() {
pIncomingMsg->Release(); pIncomingMsg->Release();
} }
} }
// Adaptive polling: if messages received, poll immediately; otherwise increase interval
if (totalMessages > 0) {
currentPollInterval_ = 0; // 有消息,立即轮询
} else {
// 无消息逐渐增加间隔最大10ms
currentPollInterval_ = std::min(currentPollInterval_ + 1, 10);
} }
// Schedule next poll
timer_->expires_after(std::chrono::milliseconds(currentPollInterval_));
timer_->async_wait([this](const boost::system::error_code& error) {
if (!error && running_) {
startAsyncPoll();
}
});
}

View File

@@ -22,8 +22,7 @@ public:
std::shared_ptr<MultiplexManager> getMultiplexManager(HSteamNetConnection conn); std::shared_ptr<MultiplexManager> getMultiplexManager(HSteamNetConnection conn);
private: private:
void run(); void startAsyncPoll();
void pollMessages();
boost::asio::io_context& io_context_; boost::asio::io_context& io_context_;
ISteamNetworkingSockets* m_pInterface_; ISteamNetworkingSockets* m_pInterface_;
@@ -34,9 +33,9 @@ private:
std::map<HSteamNetConnection, std::shared_ptr<MultiplexManager>> multiplexManagers_; std::map<HSteamNetConnection, std::shared_ptr<MultiplexManager>> multiplexManagers_;
std::thread thread_; std::unique_ptr<boost::asio::steady_timer> timer_;
std::thread io_thread_;
bool running_; bool running_;
int currentPollInterval_; // 当前轮询间隔(毫秒)
}; };
#endif // STEAM_MESSAGE_HANDLER_H #endif // STEAM_MESSAGE_HANDLER_H