Reactor模型高并发服务器——04_连接管理模块

✨✨欢迎来到T_X_Parallel的博客！！     🛰️博客主页：T_X_Parallel     🛰️项目代码仓库：Reactor模型高并发服务器项目代码仓库（随博客更新）     🛰️专栏 : Reactor模型高并发服务器项目     🛰️欢迎关注：👍点赞🙌收藏✍️留言

文章目录

• • 1. 前言
  • 2. 模块分析
  • 3. 模块实现
  • • 模块接口
    • 模块实现细节
  • 4. 模块测试
  • 5. 模块代码

项目环境：vscode、wsl2（Ubuntu22.04）

技术栈：C/C++、C++11、C++17、STL、HTTP、TCP、HTML

1. 前言

连接管理模块不名思意就是一个管理连接的模块，该模块的主要功能就是管理一个客户端连接的从连接到释放的所有过程，这个模块同样整合了前面的全部模块，包括会使用到Any类，具体的设计会在下面进行详细分析，该模块是该项目的核心模块之一，所以需要注意一些细节，毕竟整个了多个模块，也得先确保前面模块的正确，不然出错会很麻烦，这时候就考验你的调试能力了

2. 模块分析

首先先来分析该模块所需要的成员，作为一个管理客户端连接的模块，那么可以给每个连接设置一个连接ID用来区分，同时这个ID还可以当作是超时连接释放任务的定时任务ID，因为一个连接只存在一个这个定时任务，所以不需要再多使用一个变量来存这个定时任务ID；然后就是每个连接都有一个连接描述符fd和对应的Socket对象，因为这个模块会实现连接过程中所有事件的处理函数，那么必然要用到连接描述符来操作；再然后就是一个超时连接释放启用标识符；还有一个当前连接是个什么状态，即连接状态，分为连接中、连接成功、释放中、释放成功，这四个过程所需要处理的事件在模块实现部分会详细阐述；然后就是一个对应的EventLoop对象指针和连接描述符对应的Channel对象，用来设置监控连接的各种事件，还有一个存储上下文的any类，这个会在实现Http协议支持时用到，简单来说就是一个存储一个连接的一个请求响应的上下文内容的any类；同时还有两个buffer缓冲区，用来存放发送与接收数据；最后就是五个回调函数，与channel不同的是，这五个回调函数是用来处理请求与响应等事务处理函数，是应用层协议支持中传进来的，tcp层是不知道应用层需要怎么处理，这五个回调函数包括通信回调函数（处理通信信息）、连接回调函数（处理连接中到连接成功的事务）、关闭回调函数（处理连接关闭前的事务）、任意事件回调函数（任意事件触发后执行的）、服务端关闭回调函数（这个回调函数会在下面详细阐述作用，该接口关系到后面加入多线程后的程序设计）

然后再来分析该模块主要要实现的功能，成员中存在多个服务端设置的内容和回调函数，所以需要实现设置这些回调函数以及内容的功能，然后就是获取这些内容的变量的功能，比如上下文any类和连接描述符，还有业务处理的整体设置，最后就是发送响应、关闭连接、关闭连接前的一些必要检查、连接成功前一些必要设置、启动和取消超时连接释放功能，当然实现的时候需要对几个功能进行封装，有些功能存在线程安全问题，需要将这些功能的执行放入任务队列中

大致功能就这几个，具体的接口实现放在下面实现中，中间有很多细节需要注意

3. 模块实现

模块接口

公有接口

• 获取连接描述符接口
• 获取连接ID接口
• 判断是否连接成功接口
• 设置、获取上下文接口
• 设置连接、通信、连接关闭、任意事件、服务端的连接关闭业务处理回调函数
• 多个将私有接口执行的任务放入任务队列的接口，如发送响应、释放连接等

// 获取连接描述符
int GetFd();

// 获取连接ID
int GetConnId();

// 判断是否连接成功
bool IsConnected();

// 设置上下文
void SetContext(const std::any &context);

// 获取上下文
std::any *GetContext();

// 设置连接业务处理回调函数
void SetConnectedCallBack(const ConnectedCallBack &cb);

// 设置通信业务处理回调函数
void SetMessageCallBack(const MessageCallBack &cb);

// 设置连接关闭业务处理回调函数
void SetClosedCallBack(const ClosedCallBack &cb);

// 设置任意事件触发业务处理回调函数
void SetAnyEventCallBack(const AnyEventCallBack &cb);

// 设置服务端设置的连接关闭业务处理回调函数
void SetServerClosedCallBack(const ClosedCallBack &cb);

// 下面是对上面的私有接口封装，上面的几个私有接口可能存在线程安全问题，需放入eventloop中的任务队列中

void Established();

void Send(const char *ch, size_t len);

void ShutDown();

void Release();

void EnableInactiveRelease(int sec);

void CancelInactiveRelease();

void UpdatedProtocol(const std::any &context,
const ConnectedCallBack &concb, const MessageCallBack &mescb,
const ClosedCallBack &clocb, const AnyEventCallBack &anecb);

私有接口

• 可写、可读、错误、关闭、任意事件处理函数
• 连接成功前业务处理函数
• 连接释放函数（真正释放）
• 发送响应给客户端函数
• 连接释放前检查函数
• 启动、取消超时连接释放
• 更新协议对应的上下文处理以及连接业务处理回调函数的函数

// 可写事件处理函数
void HandleWrite();

// 可读事件处理函数
void HandleRead();
// 关闭事件处理函数
void HandleClose();

// 错误事件处理函数
void HandleError();

// 任意事件处理函数
void HandleAnyEvent();

// 连接成功前业务处理函数
void _Established();

// 连接释放函数
void _Release();

// 发送响应给客户端
void _Send(const std::string &s);

// 连接释放前检查
void _ShutDown();

// 启动超时连接释放
void _EnableInactiveRelease(int sec);

// 取消超时连接释放
void _CancelInactiveRelease();

// 更新协议对应的上下文处理以及连接业务处理回调函数
void _UpdatedProtocol(const std::any &context,
const ConnectedCallBack &concb, const MessageCallBack &mescb,
const ClosedCallBack &clocb, const AnyEventCallBack &anecb);

模块实现细节

可读事件处理函数中，过程就是从连接描述符中读出数据放在自己实现的接收缓冲区中，然后执行通信业务处理回调函数处理接收到的数据，其中在从内核缓冲区中读数据时，之前设计过当对方关闭连接时会返回-1，那么只要返回-1，则我们可以先做释放前检查再释放连接

可写事件处理函数中，过程就是从发送缓冲区中读出处理完的响应数据，然后发送出去，在发送后的返回值表示出现错误，那么可以直接释放，在释放前需检查接收缓冲区中是否有数据，有数据则先执行业务处理再释放，写完之后如果发送缓冲区中无数据则关闭读事件监控，并判断目前是否处于连接关闭中状态，如果是，则直接调用接口释放连接

服务端设置的关闭连接回调函数的作用是，在后面加入多线程后，在上层服务端中，需要管理多个连接，那么关闭连接就由上层服务端决定，所以上层服务端存储的是每个连接的智能指针share_ptr，只有上层服务端中的智能指针释放了，该连接才能算真正释放，所以服务端传入了一个释放掉他那里的指针的回调函数，具体作用体现在实现tcpsever模块中就会明了

发送响应接口中要做的事就是将处理好的响应放入发送缓冲区中，但是传入的是一个字符串指针，说明在写入缓冲区的时候该字符串指针的内容可能会被释放，所以在这个地方需要先将该字符串指针指向的内容存放在一个临时变量中

在外部传入的业务处理回调函数的参数中是Connection对象的智能指针的引用，而在类中需要传入，这时候需要将该类继承于std::enable_shared_from_this<Connection>来获取智能指针，需要获取的时候调用shared_from_this()即可

最后一个实现细节就是真正释放接口是把该任务直接放入任务队列中而不是先判断是否在对应进程，是则直接执行，不是再放入任务队列，这里是为了让任务队列中的任务执行完再释放，否则会出现问题

这些细节注意后其他的实现应该没有什么问题，具体的实现请看完整代码进行分析，如对哪里存在疑惑，请在评论区留言

4. 模块测试

测试代码

// 服务端
std::unordered_map<uint64_t, ns_connection::SharePtrConnection> conns;
uint64_t id = 1;
void OnConnected(const ns_connection::SharePtrConnection &conn)
{
LOG(NORMAL, "New Connection: " + std::to_string(conn->GetFd()));
}

void OnClose(const ns_connection::SharePtrConnection &conn)
{
LOG(NORMAL, "Close Connection: " + std::to_string(conn->GetFd()));
conns.erase(conn->GetConnId());
}

void OnMessage(const ns_connection::SharePtrConnection &conn, ns_buffer::Buffer *buff)
{
std::string s = buff->Read();
LOG(NORMAL, s);
s = "xiaomi yu7";
conn->Send(s.c_str(), s.size());
}

void Accept(ns_eventloop::EventLoop *loop, ns_eventloop::Channel *ls_ch)
{
int fd = ls_ch->GetFd();
int newfd = accept(fd, NULL, NULL);
// int newfd = svr_socket.Accept();
if (newfd < 0)
{
LOG(DEBUG, "accept failed");
return;
}
ns_connection::SharePtrConnection conn(new ns_connection::Connection(id, newfd, loop));
conn->SetConnectedCallBack(std::bind(OnConnected, std::placeholders::_1));
conn->SetMessageCallBack(std::bind(OnMessage, std::placeholders::_1, std::placeholders::_2));
conn->SetServerClosedCallBack(std::bind(OnClose, std::placeholders::_1));
conn->EnableInactiveRelease(10);
conn->Established();
conns.insert(std::make_pair(id++, conn));
}

int main()
{
ns_eventloop::EventLoop loop;
ns_socket::Socket svr_socket;
svr_socket.CreateServer(8085);
ns_eventloop::Channel channel(svr_socket.GetFd(), &loop);
channel.SetReadCallBack(std::bind(&Accept, &loop, &channel));
channel.EnableRead();
loop.Start();
svr_socket.Close();
return 0;
}
// 客户端
int main()
{
ns_socket::Socket cli_socket;
cli_socket.CreateClient(8085, "127.0.0.1");

for (int i = 0; i < 10; i++)
{
std::string str = "xiaomi su7ultra";
cli_socket.Send(str.c_str(), str.size());
char buf[1024] = {0};
cli_socket.Recv(&buf, 1023);
LOG(NORMAL, buf);
sleep(1);
}
int i = 1;
while(1)
sleep(1);
cli_socket.Close();
return 0;
}

注：由于该模块使用Buffer模块和any类，所以需要C++17标准，否则会编译报错

测试结果

// 服务端
./Server
[Level# NORMAL][time# 2025–04–05 15:51:02][File# test_connection_server.cc][Line# 9]Message# New Connection: 7
[Level# NORMAL][time# 2025–04–05 15:51:02][File# test_connection_server.cc][Line# 21]Message# xiaomi su7ultra
[Level# NORMAL][time# 2025–04–05 15:51:03][File# test_connection_server.cc][Line# 21]Message# xiaomi su7ultra
[Level# NORMAL][time# 2025–04–05 15:51:04][File# test_connection_server.cc][Line# 21]Message# xiaomi su7ultra
[Level# NORMAL][time# 2025–04–05 15:51:05][File# test_connection_server.cc][Line# 21]Message# xiaomi su7ultra
[Level# NORMAL][time# 2025–04–05 15:51:06][File# test_connection_server.cc][Line# 21]Message# xiaomi su7ultra
[Level# NORMAL][time# 2025–04–05 15:51:07][File# test_connection_server.cc][Line# 21]Message# xiaomi su7ultra
[Level# NORMAL][time# 2025–04–05 15:51:08][File# test_connection_server.cc][Line# 21]Message# xiaomi su7ultra
[Level# NORMAL][time# 2025–04–05 15:51:09][File# test_connection_server.cc][Line# 21]Message# xiaomi su7ultra
[Level# NORMAL][time# 2025–04–05 15:51:10][File# test_connection_server.cc][Line# 21]Message# xiaomi su7ultra
[Level# NORMAL][time# 2025–04–05 15:51:11][File# test_connection_server.cc][Line# 21]Message# xiaomi su7ultra
[Level# NORMAL][time# 2025–04–05 15:51:20][File# test_connection_server.cc][Line# 14]Message# Close Connection: 7

// 客户端
./Client
[Level# NORMAL][time# 2025–04–05 15:51:02][File# test_connection_client.cc][Line# 29]Message# xiaomi yu7
[Level# NORMAL][time# 2025–04–05 15:51:03][File# test_connection_client.cc][Line# 29]Message# xiaomi yu7
[Level# NORMAL][time# 2025–04–05 15:51:04][File# test_connection_client.cc][Line# 29]Message# xiaomi yu7
[Level# NORMAL][time# 2025–04–05 15:51:05][File# test_connection_client.cc][Line# 29]Message# xiaomi yu7
[Level# NORMAL][time# 2025–04–05 15:51:06][File# test_connection_client.cc][Line# 29]Message# xiaomi yu7
[Level# NORMAL][time# 2025–04–05 15:51:07][File# test_connection_client.cc][Line# 29]Message# xiaomi yu7
[Level# NORMAL][time# 2025–04–05 15:51:08][File# test_connection_client.cc][Line# 29]Message# xiaomi yu7
[Level# NORMAL][time# 2025–04–05 15:51:09][File# test_connection_client.cc][Line# 29]Message# xiaomi yu7
[Level# NORMAL][time# 2025–04–05 15:51:10][File# test_connection_client.cc][Line# 29]Message# xiaomi yu7
[Level# NORMAL][time# 2025–04–05 15:51:11][File# test_connection_client.cc][Line# 29]Message# xiaomi yu7

测试结果中显示能正常通信，也能正常超时连接释放，可以自行尝试启动多个客户端进行连接

5. 模块代码

// 连接管理模块代码
#pragma once

#include <iostream>
#include <any>
#include "log.hpp"
#include "eventloop.hpp"
#include "buffer.hpp"
#include "socket.hpp"

namespace ns_connection
{
class Connection;

typedef enum
{
CONNECTING,
CONNECTED,
DISCONNECTING,
DISCONNECTED
} ConnectStatus;

typedef std::shared_ptr<Connection> SharePtrConnection;

typedef std::function<void(const SharePtrConnection &)> ConnectedCallBack;
typedef std::function<void(const SharePtrConnection &, ns_buffer::Buffer *)> MessageCallBack;
typedef std::function<void(const SharePtrConnection &)> ClosedCallBack;
typedef std::function<void(const SharePtrConnection &)> AnyEventCallBack;

class Connection : public std::enable_shared_from_this<Connection>
{
private:
uint64_t _connection_id;
int _socket_fd;
bool _inactive_release_sign;
ConnectStatus _status;

ns_eventloop::EventLoop *_loop;
ns_eventloop::Channel _channel;
ns_socket::Socket _socket;
ns_buffer::Buffer _in_buff;
ns_buffer::Buffer _out_buff;
std::any _context;

ConnectedCallBack _connected_callback;
MessageCallBack _message_callback;
ClosedCallBack _closed_callback;
AnyEventCallBack _anyevent_callback;
ClosedCallBack _server_closed_callback;

private:
// 可写事件处理函数
void HandleWrite()
{
std::string s = _out_buff.Read();
ssize_t ret = _socket.Send(s.c_str(), s.size(), true);
if (ret < 0)
{
if (_in_buff.AbleRead())
_message_callback(shared_from_this(), &_in_buff);
Release();
return;
}
if (!_out_buff.AbleRead())
{
_channel.DisableWrite();
if (_status == DISCONNECTING)
Release();
}
}

// 可读事件处理函数
void HandleRead()
{
char buff[65536];
ssize_t ret = _socket.Recv(buff, 65536, true);
if (ret < 0)
{
LOG(WARNING, "The Other Party Closes The Connection");
// ShutDown();
_ShutDown();
return;
}
_in_buff.Write(buff, ret);
if (_in_buff.AbleRead())
if (_message_callback)
_message_callback(shared_from_this(), &_in_buff);
}

// 关闭事件处理函数
void HandleClose()
{
if (_in_buff.AbleRead())
_message_callback(shared_from_this(), &_in_buff);
Release();
}

// 错误事件处理函数
void HandleError()
{
if (_in_buff.AbleRead())
_message_callback(shared_from_this(), &_in_buff);
Release();
}

// 任意事件处理函数
void HandleAnyEvent()
{
if (_inactive_release_sign == true)
_loop->RefreshTimerTask(_connection_id);
if (_anyevent_callback)
_anyevent_callback(shared_from_this());
}

// 连接成功前业务处理函数
void _Established()
{
if (_status != CONNECTING)
{
LOG(ERROR, "Status isn't CONNECTING");
return;
}
_status = CONNECTED;
_channel.EnableRead();
if (_connected_callback)
_connected_callback(shared_from_this());
}

// 连接释放函数
void _Release()
{
if (_status == DISCONNECTED)
{
LOG(WARNING, "Connection Dropped");
return;
}
_status = DISCONNECTED;
_channel.Remove();
_socket.Close();
if (_loop->ExistTimerTask(_connection_id))
_loop->CancelTimerTask(_connection_id);
if (_closed_callback)
_closed_callback(shared_from_this());
if (_server_closed_callback)
_server_closed_callback(shared_from_this());
}

// 发送响应给客户端
void _Send(const std::string &s)
{
if (_status == DISCONNECTED)
{
LOG(WARNING, "Connection Dropped");
return;
}
_out_buff.Write(s);
if (_channel.IsWriteEvent() == false)
_channel.EnableWrite();
}

// 连接释放前检查
void _ShutDown()
{
_status = DISCONNECTING;
if (_in_buff.AbleRead())
if (_message_callback)
_message_callback(shared_from_this(), &_in_buff);
if (_out_buff.AbleRead())
if (_channel.IsWriteEvent() == false)
_channel.EnableWrite();
if (!_out_buff.AbleRead() && !_in_buff.AbleRead())
Release();
}

// 启动超时连接释放
void _EnableInactiveRelease(int sec)
{
_inactive_release_sign = true;
if (_loop->ExistTimerTask(_connection_id))
_loop->RefreshTimerTask(_connection_id);
else
_loop->AddTimerTask(_connection_id, sec, std::bind(&Connection::Release, this));
}

// 取消超时连接释放
void _CancelInactiveRelease()
{
_inactive_release_sign = false;
if (_loop->ExistTimerTask(_connection_id))
_loop->CancelTimerTask(_connection_id);
}

// 更新协议对应的上下文处理以及连接业务处理回调函数
void _UpdatedProtocol(const std::any &context, const ConnectedCallBack &concb, const MessageCallBack &mescb, const ClosedCallBack &clocb, const AnyEventCallBack &anecb)
{
_context = context;
_connected_callback = concb;
_message_callback = mescb;
_closed_callback = clocb;
_anyevent_callback = anecb;
}

public:
Connection(uint64_t connection_id, int sockfd, ns_eventloop::EventLoop *loop)
: _connection_id(connection_id), _socket_fd(sockfd), _inactive_release_sign(false), _status(CONNECTING), _socket(_socket_fd), _loop(loop), _channel(_socket_fd, loop)
{
_channel.SetWriteCallBack(std::bind(&Connection::HandleWrite, this));
_channel.SetReadCallBack(std::bind(&Connection::HandleRead, this));
_channel.SetCloseCallBack(std::bind(&Connection::HandleClose, this));
_channel.SetErrorCallBack(std::bind(&Connection::HandleError, this));
_channel.SetAnyEventCallBack(std::bind(&Connection::HandleAnyEvent, this));
}

// ~Connection()
// {
// LOG(DEBUG, "Release Address: " + std::to_string((uintptr_t)this));
// }

// 获取连接描述符
int GetFd()
{
return _socket_fd;
}

// 获取连接ID
int GetConnId()
{
return _connection_id;
}

// 判断是否连接成功
bool IsConnected()
{
return _status == CONNECTED;
}

// 设置上下文
void SetContext(const std::any &context)
{
_context = context;
}

// 获取上下文
std::any *GetContext()
{
return &_context;
}

// 设置连接业务处理回调函数
void SetConnectedCallBack(const ConnectedCallBack &cb)
{
_connected_callback = cb;
}

// 设置通信业务处理回调函数
void SetMessageCallBack(const MessageCallBack &cb)
{
_message_callback = cb;
}

// 设置连接关闭业务处理回调函数
void SetClosedCallBack(const ClosedCallBack &cb)
{
_closed_callback = cb;
}

// 设置任意事件触发业务处理回调函数
void SetAnyEventCallBack(const AnyEventCallBack &cb)
{
_anyevent_callback = cb;
}
// 设置服务端设置的连接关闭业务处理回调函数
void SetServerClosedCallBack(const ClosedCallBack &cb)
{
_server_closed_callback = cb;
}

// 下面是对上面的私有接口封装，上面的几个私有接口可能存在线程安全问题，需放入eventloop中的任务队列中

void Established()
{
_loop->RunInLoop(std::bind(&Connection::_Established, this));
}

void Send(const char *ch, size_t len)
{
std::string s(ch, len);
_loop->RunInLoop(std::bind(&Connection::_Send, this, s));
}

void ShutDown()
{
_loop->RunInLoop(std::bind(&Connection::_ShutDown, this));
}

void Release()
{
// _loop->RunInLoop(std::bind(&Connection::_Release, this));
// 直接放入任务队列中是因为需要先将所有任务队列中的任务执行完再释放连接
_loop->InsertTaskPool(std::bind(&Connection::_Release, this));
}

void EnableInactiveRelease(int sec)
{
_loop->RunInLoop(std::bind(&Connection::_EnableInactiveRelease, this, sec));
}

void CancelInactiveRelease()
{
_loop->RunInLoop(std::bind(&Connection::_CancelInactiveRelease, this));
}

void UpdatedProtocol(const std::any &context, const ConnectedCallBack &concb, const MessageCallBack &mescb, const ClosedCallBack &clocb, const AnyEventCallBack &anecb)
{
_loop->AssertInLoop();
_loop->RunInLoop(bind(&Connection::_UpdatedProtocol, this, context, concb, mescb, clocb, anecb));
}
};
} // namespace ns_connection

下个LoopThread模块和Acceptor模块的实现就正式加入多线程，而且后面的获取除了HttpServer模块之外实现的逻辑都比较简单

专栏：Reactor模型高并发服务器项目 项目代码仓库：Reactor模型高并发服务器项目代码仓库（随博客更新） 都看到这里了，留下你们的珍贵的👍点赞+⭐收藏+📋评论吧