add:添加无心跳帧客户端的超时踢出(目前是10分钟)。

This commit is contained in:
taynpg 2025-01-06 14:39:53 +08:00
parent 5d18697510
commit d82a65dea6
3 changed files with 52 additions and 7 deletions

View File

@ -134,6 +134,10 @@
"concepts": "cpp",
"coroutine": "cpp",
"format": "cpp",
"stop_token": "cpp"
"stop_token": "cpp",
"expected": "cpp",
"numbers": "cpp",
"semaphore": "cpp",
"span": "cpp"
}
}

View File

@ -4,15 +4,22 @@
using namespace ofen;
constexpr int g_MaxCacheLen = 1024 * 1024 * 50;
constexpr int check_idle_percycle = 1000 * 60; // 毫秒
constexpr int remove_after_time = 60 * 10; // 秒
CTcpServer::CTcpServer(asio::io_context& io_context, const std::shared_ptr<spdlog::logger>& logger)
: io_context_(io_context), acceptor_(io_context), logger_(logger)
{
th_run_ = true;
sleep_.set_timeout(check_idle_percycle);
}
CTcpServer::~CTcpServer()
{
th_run_ = false;
sleep_.contiune();
if (th_monitor_idle_.joinable()) {
th_monitor_idle_.join();
}
}
bool CTcpServer::start(unsigned short port)
@ -30,6 +37,7 @@ bool CTcpServer::start(unsigned short port)
auto bound_endpoint = acceptor_.local_endpoint();
server_ip_ = bound_endpoint.address().to_string() + ":" + std::to_string(bound_endpoint.port());
accept_client();
th_monitor_idle_ = std::thread([this]() { monitor_idle(); });
logger_->info("Server started on port {}", port);
return true;
}
@ -201,6 +209,7 @@ void CTcpServer::accept_client()
auto cache = std::make_shared<ClientCache>();
cache->socket_ = socket;
cache->online_time_ = OfUtil::now_time();
cache->last_active_time_ = std::chrono::high_resolution_clock::now();
client_map_[client_key] = cache;
can = true;
}
@ -244,7 +253,6 @@ void CTcpServer::th_client(std::shared_ptr<asio::ip::tcp::socket> socket, const
asio::error_code error;
size_t length = socket->read_some(asio::buffer(cache->tmp_buf_), error);
if (error == asio::error::eof) {
logger_->info("Connection closed by client: {}", client_key);
break;
} else if (error) {
throw asio::system_error(error);
@ -255,6 +263,11 @@ void CTcpServer::th_client(std::shared_ptr<asio::ip::tcp::socket> socket, const
auto* frame = CTransProtocal::parse(cache->buffer_);
if (frame) {
if (frame->type_ == TYPE_HEARTS) {
std::lock_guard<std::mutex> lock(cli_mut_);
if (client_map_.count(client_key)) {
auto& cli = client_map_[client_key];
cli->last_active_time_ = std::chrono::high_resolution_clock::now();
}
delete frame;
continue;
}
@ -299,3 +312,30 @@ bool CTcpServer::send_frame(std::shared_ptr<asio::ip::tcp::socket> socket, CFram
delete[] out_buf;
return true;
}
void CTcpServer::monitor_idle()
{
while (th_run_) {
sleep_.sleep();
if (!th_run_) {
break;
}
std::vector<std::string> remove_vec;
std::lock_guard<std::mutex> lock(cli_mut_);
for (auto& item : client_map_) {
auto now = std::chrono::high_resolution_clock::now();
auto duration =
std::chrono::duration_cast<std::chrono::seconds>(now - item.second->last_active_time_)
.count();
if (duration >= remove_after_time) {
logger_->warn("OnLine Time [{}] sec, Proactively disconnect:{}", duration, item.first);
remove_vec.push_back(item.first);
item.second->socket_->shutdown(asio::ip::tcp::socket::shutdown_both);
item.second->socket_->close();
}
}
for (const auto& item : remove_vec) {
client_map_.erase(item);
}
}
}

View File

@ -5,8 +5,10 @@
#include <string>
#include <util.h>
#include <vector>
#include <chrono>
using namespace ofen;
using high_c = std::chrono::time_point<std::chrono::high_resolution_clock>;
struct ClientCache {
std::shared_ptr<asio::ip::tcp::socket> socket_;
CMutBuffer buffer_{};
@ -14,6 +16,7 @@ struct ClientCache {
std::string task_{};
std::string task_time_{};
std::string online_time_{};
high_c last_active_time_;
FrameType cur_type_{TYPE_DEFAULT};
};
struct TaskList {
@ -45,12 +48,8 @@ private:
private:
void accept_client();
void th_client(std::shared_ptr<asio::ip::tcp::socket> socket, const std::string& client_key);
/// @brief 不删除 buf
/// @param socket
/// @param buf
/// @return
bool send_frame(std::shared_ptr<asio::ip::tcp::socket> socket, CFrameBuffer* buf);
void monitor_idle();
private:
bool th_run_{false};
@ -61,5 +60,7 @@ private:
std::map<std::string, std::thread> client_threads_;
std::mutex cli_mut_;
std::mutex buf_mut_;
std::thread th_monitor_idle_;
std::string server_ip_;
CThreadSleep sleep_;
};