【Linux高级IO（三）】Reactor - IT评测·应用市场-qidao123.com技术社区

#pragma once
#include <iostream>
#include <string>
#include <memory>
#include <cerrno>
#include <functional>
#include <unordered_map>
#include "Log.hpp"
#include "nocopy.hpp"
#include "Epoller.hpp"
#include "Socket.hpp"
#include "Comm.hpp"
class Connection;//代表一个客户端的连接，存储与该连接相关的信息与回调函数
class TcpServer;//实现了 TCP 服务器的核心功能，包含监听套接字、epoll 实例以及管理所有连接的映射。
uint32_t EVENT_IN = (EPOLLIN | EPOLLET); //读ET
uint32_t EVENT_OUT = (EPOLLOUT | EPOLLET); //写ET
const static int g_buffer_size = 128;
//func_t表示一个函数对象类型函数接收一个指向Connection 对象的弱指针std::weak_ptr<Connection> 作为参数
using func_t = std::function<void(std::weak_ptr<Connection>)>;
using except_func = std::function<void(std::weak_ptr<Connection>)>;
class Connection
{
public:
Connection(int sock) : _sock(sock)
{
}
void SetHandler(func_t recv_cb, func_t send_cb, except_func except_cb)
{
_recv_cb = recv_cb;
_send_cb = send_cb;
_except_cb = except_cb;
}
int SockFd() { return _sock; }
void AppendInBuffer(const std::string &info)//向输入缓冲区中追加数据
{
_inbuffer += info; //传入的字符串追加到输入缓冲区中
}
void AppendOutBuffer(const std::string &info)
{
_outbuffer += info;
}
std::string &Inbuffer() // for debug
{
return _inbuffer;//返回输入缓冲区_inbuffer的引用方便爱不访问和查看输入缓冲区的内容
}
std::string &OutBuffer()
{
return _outbuffer;
}
//设置指向TcpServer对象的弱指针_tcp_server_ptr
void SetWeakPtr(std::weak_ptr<TcpServer> tcp_server_ptr)
{
_tcp_server_ptr = tcp_server_ptr;
}
~Connection()
{
}
private:
int _sock;
std::string _inbuffer; // string 二进制流，vector
std::string _outbuffer;
public:
func_t _recv_cb;//接收回调函数
func_t _send_cb;//发送回调函数
except_func _except_cb;//异常处理回调函数
// 添加一个回指指针
std::weak_ptr<TcpServer> _tcp_server_ptr; // std::weak_ptr<> // bug??
// std::shared_ptr<TcpServer> _tcp_server_ptr; // std::weak_ptr<>
std::string _ip;
uint16_t _port;
};
// enable_shared_from_this:可以提供返回当前对象的this对应的shared_ptr
class TcpServer : public std::enable_shared_from_this<TcpServer>, public nocopy
{
static const int num = 64;
public:
TcpServer(uint16_t port, func_t OnMessage)
: _port(port),
_OnMessage(OnMessage),
_quit(true),
_epoller_ptr(new Epoller()),
_listensock_ptr(new Sock())
{
}
void Init() //创建监听套接字将其设为非阻塞并将它添加到epoll实例中
{
_listensock_ptr->Socket();
SetNonBlockOrDie(_listensock_ptr->Fd());
_listensock_ptr->Bind(_port);
_listensock_ptr->Listen();
lg(Info, "create listen socket success: %d", _listensock_ptr->Fd());
//将监听套接字添加到epoll中，设置监听读事件（EVENT_IN）并绑定Accepter函数作为该事件回调函数，同时设置数据发送和异常处理回调函数为nullptr（因为监听套接字主要用于接受连接）
// ？？？？？？？？？？？？？？？？？？？？？？？？
//将Accepter函数作为一个回调函数绑定并注册，一般在特定的事件发生时调用，而不是在AddConnection 函数调用时就执行
//通过 AddConnection 函数将 Accepter 函数作为回调函数注册到监听套接字的 EVENT_IN事件上
//意味着当有新客户端连接请求到达时，（即监听套接字上发生 EVENT_IN 事件），系统会自动调用Accepter 函数来处理这个新连接
//获取文件描述符，添加到epoll中，AddConnection并告诉epoll关注它的EVENT_IN 事件
//AddConnection 函数会将 Accepter 函数作为回调函数与 EVENT_IN 事件绑定。
//当事件监听机制检测到监听套接字上发生 EVENT_IN 事件时，会调用之前绑定的 Accepter 函数来处理这个新的客户端连接请求。
AddConnection(_listensock_ptr->Fd(),
EVENT_IN, std::bind(&TcpServer::Accepter, this, std::placeholders::_1), nullptr, nullptr);
}
//创建新的Connection对象，将其添加到_connections映射中，将对应的事件添加到epoll实例中
void AddConnection(int sock, uint32_t event, func_t recv_cb, func_t send_cb, except_func except_cb,
const std::string &ip = "0.0.0.0", uint16_t port = 0)
{
// 1. 给sock也建立一个connection对象，将listensock添加到Connection中，同时，listensock和Connecion放入_connections
// std::shared_ptr<Connection> new_connection = std::make_shared<Connection>(sock, std::shared_ptr<TcpServer>(this));
// std::shared_ptr<Connection> new_connection = std::make_shared<Connection>(sock);
//创建新的Connection对象 new_connection，使用传入的套接字描述符sock进行舒适化使用new关键字创建并封装为std::shared_ptr
std::shared_ptr<Connection> new_connection(new Connection(sock));
new_connection->SetWeakPtr(shared_from_this()); // shared_from_this(): 返回当前对象的shared_ptr //传入当前TcpServer对象的std::shared_ptr 设置指向 TcpServer 的弱指针
new_connection->SetHandler(recv_cb, send_cb, except_cb);//设置回调函数把传入的三个回调函数设置到Connection对象中，以便在？？？相应事件？？？？（读、写、异常）发生时调用这些回调函数进行处理
new_connection->_ip = ip;
new_connection->_port = port;
// // 2. 添加到unordered_map
_connections.insert(std::make_pair(sock, new_connection));
// // 3. 我们添加对应的事件，除了要加到内核中，fd， event
_epoller_ptr->EpollerUpdate(EPOLL_CTL_ADD, sock, event);
// lg(Debug, "add a new connection success, sockfd is : %d", sock);
}
// 链接管理器
//处理新的客户端连接接受连接并将其添加到epoll实例中
void Accepter(std::weak_ptr<Connection> conn)
{
auto connection = conn.lock();
while (true)
{
struct sockaddr_in peer;
socklen_t len = sizeof(peer);
//系统调用accept函数接受一个新的客户端连接获取监听套接字的文件描述符
int sock = ::accept(connection->SockFd(), (struct sockaddr *)&peer, &len);
if (sock > 0) //成功接受了一个新的客户端连接
{
uint16_t peerport = ntohs(peer.sin_port);
char ipbuf[128];
inet_ntop(AF_INET, &peer.sin_addr.s_addr, ipbuf, sizeof(ipbuf));
lg(Debug, "get a new client, get info-> [%s:%d], sockfd : %d", ipbuf, peerport, sock);
SetNonBlockOrDie(sock);
// listensock只需要设置_recv_cb, 而其他sock，读，写，异常
//AddConnection 函数工作---- 创建Connection 对象，设置相关的回调函数，将连接信息存储到 _connections 映射中，
//并将套接字和事件添加到 epoll 实例中进行监听
AddConnection(sock, EVENT_IN,
std::bind(&TcpServer::Recver, this, std::placeholders::_1),
//将数据接收的逻辑封装在Recver函数中，使数据接收的处理和Accepter函数的连接接受逻辑分离。这样可以提高代码模块化
//&TcpServer::Recver 是TcpServer类里的Recver成员函数 this代表当前TcpServer对象的指针
//std::placeholders::_1 是占位符，意味着调用这个绑定函数时会传入一个参数。
//当 EVENT_IN 事件触发（即套接字有数据可读）时，AddConnection 函数会调用这个绑定的 Recver 函数来处理接收到的数据。
std::bind(&TcpServer::Sender, this, std::placeholders::_1),
//当套接字上发生写事件（EPOLLOUT）会调用这个绑定的 Sender 函数。该函数负责将数据从应用程序的输出缓冲区发送到输出缓冲区中
std::bind(&TcpServer::Excepter, this, std::placeholders::_1),
//当这个套接字上出现异常情况（例如连接断开、网络错误等）时，AddConnection 函数会调用这个绑定的 Excepter 函数来处理异常。
ipbuf, peerport); // TODO
}
else
{
if (errno == EWOULDBLOCK)
break;
else if (errno == EINTR)
continue;
else
break;
}
}
}
// 事件管理器
// 应不应该关心数据的格式？？？不应该！！服务器只要IO数据就可以，有没有读完，报文的格式细节，你不用管。
//处理数据接收，将接收到的数据添加到输入缓冲区，并调用_OnMessage回调函数
void Recver(std::weak_ptr<Connection> conn) //接收一个指向Connection的弱指针
{
if(conn.expired()) return;//检查弱指针是否过期即所指向的Connection是或否已经被释放释放就返回不进行后续操作
auto connection = conn.lock();//尝试将弱指针 conn 提升为强指针 connection。如果 conn 没有过期，lock 函数会返回一个有效的强指针，否则返回一个空指针。
int sock = connection->SockFd(); //获取当前连接的套接字描述符sock
while (true) //持续接收数据
{
char buffer[g_buffer_size];
memset(buffer, 0, sizeof(buffer));
//调用recv系统函数从sock接收数据到buffer中，buffer 大小减 1 个字节的数据（留出一个字节用于存储字符串结束符 \0）
ssize_t n = recv(sock, buffer, sizeof(buffer) - 1, 0); // 非阻塞读取
if (n > 0)
{
connection->AppendInBuffer(buffer);//将接收到数据buffer追加到连接的输入缓冲区中
}
else if (n == 0)
{
lg(Info, "sockfd: %d, client info %s:%d quit...", sock, connection->_ip.c_str(), connection->_port);
connection->_except_cb(connection);
return;
}
else //小于0 接收数据发生了错误
{
if (errno == EWOULDBLOCK) //错误码是这个表示当前没有数据可读
break;
else if (errno == EINTR) //是这个表示recv函数被信号中断，继续循环，再次尝试接收数据
continue;
else
{
lg(Warning, "sockfd: %d, client info %s:%d recv error...", sock, connection->_ip.c_str(), connection->_port);
connection->_except_cb(connection);
return;
}
}
}
// 数据有了，但是不一定全，1. 检测 2. 如果有完整报文，就处理
_OnMessage(connection); // 你读到的sock所有的数据connection
}
//处理数据发送将输出缓冲区中的数据发送给客户端，并根据发送情况调整对事件的关注
void Sender(std::weak_ptr<Connection> conn)
{
if(conn.expired()) return;
auto connection = conn.lock();
auto &outbuffer = connection->OutBuffer(); //获取连接的输出缓冲区的引用
while(true)
{
//调用send系统函数，将输出缓冲区outbuffer中的数据发送到连接的套接字connection->SockFd()上、获取缓冲区的字符数组表示、要发送的数据长度
ssize_t n = send(connection->SockFd(), outbuffer.c_str(), outbuffer.size(), 0);
if(n > 0)
{
outbuffer.erase(0, n); //从发送缓冲区的起始位置删除已经发送的n的字节的数据
if(outbuffer.empty()) break; //缓冲区中数据已经发完跳出
}
else if(n == 0)
{
return;
}
else //出现错误
{
if(errno == EWOULDBLOCK) break;
else if(errno == EINTR) continue;
else{
lg(Warning, "sockfd: %d, client info %s:%d send error...", connection->SockFd(), connection->_ip.c_str(), connection->_port);
connection->_except_cb(connection);
return;
}
}
}
if(!outbuffer.empty()) //不为空有未发送的数据
{
// 开启对写事件的关心写
//----解释----每个套接字（也是一种特殊的文件描述符）都关联两个缓冲区
//代码中，当要发送的缓冲区不为空时，即还有没发送完的数据的时候
//我们就要开启对写事件的关心
//这个写，是写到我们这个套接字对应的发送缓冲区的
//即当发送缓冲区有空间让我们写的时候，就会通知我们
//让我们进行写----写到发送缓冲区里面（send函数）
//一个套接字关注可写其实意味着，可以让应用程序及时了解套接字发送缓冲区的状态，当套接字可写时，epoll会通知程序，程序可以继续调用send函数尝试发送剩余数据
//如果不开启对些事件的关注，那么即使套接字可写程序也无法得知，导致数据发不出去
EnableEvent(connection->SockFd(), true, true);//读、写
}
else//已发送完毕
{
// 关闭对写事件的关心
//因为此时没有数据需要发送，即使套接字可写，也不需要程序进行任何操作，
//关闭写事件的关注可以减少 epoll 的事件通知开销，提高程序的运行效率。
EnableEvent(connection->SockFd(), true, false);//读、不写
}
}
//处理异常情况，从 epoll 实例中移除异常连接，关闭套接字，并从 _connections 映射中移除该连接。
void Excepter(std::weak_ptr<Connection> connection)
{
if(connection.expired()) return;
auto conn = connection.lock();
int fd = conn->SockFd();
lg(Warning, "Excepter hander sockfd: %d, client info %s:%d excepter handler",
conn->SockFd(), conn->_ip.c_str(), conn->_port);
// 1. 移除对特定fd的关心
// EnableEvent(connection->SockFd(), false, false);
_epoller_ptr->EpollerUpdate(EPOLL_CTL_DEL, fd, 0);
// 2. 关闭异常的文件描述符
lg(Debug, "close %d done...\n", fd);
close(fd);
// 3. 从unordered_map中移除
lg(Debug, "remove %d from _connections...\n", fd);
// TODO bug
// auto iter = _connections.find(fd);
// if(iter == _connections.end()) return;
// _connections.erase(iter);
// _connections[fd].reset();
_connections.erase(fd);
}
//调整对指定套接字的读写事件的关注。
void EnableEvent(int sock, bool readable, bool writeable)
{
uint32_t events = 0;
events |= ((readable ? EPOLLIN : 0) | (writeable ? EPOLLOUT : 0) | EPOLLET);
_epoller_ptr->EpollerUpdate(EPOLL_CTL_MOD, sock, events);
}
//检查指定的套接字是否存在于 _connections 映射中。
bool IsConnectionSafe(int fd)
{
auto iter = _connections.find(fd);
if (iter == _connections.end()) //走到结尾了说明没找到
return false;
else
return true; //找到了
}
//处理 epoll_wait 返回的事件，根据事件类型调用相应的回调函数。
void Dispatcher(int timeout)
{
int n = _epoller_ptr->EpollerWait(revs, num, timeout);
for (int i = 0; i < n; i++)
{
uint32_t events = revs[i].events;
int sock = revs[i].data.fd;
// 统一把事件异常转换成为读写问题
// if (events & EPOLLERR)
// events |= (EPOLLIN | EPOLLOUT);
// if (events & EPOLLHUP)
// events |= (EPOLLIN | EPOLLOUT);
// 只需要处理EPOLLIN EPOLLOUT
if ((events & EPOLLIN) && IsConnectionSafe(sock)) //事件包含读事件、并且文件描述符对应的连接存在于_connections中
{
if (_connections[sock]->_recv_cb)
_connections[sock]->_recv_cb(_connections[sock]); //连接的接收回调函数 //？？？？为什么传入这个参数
}
if ((events & EPOLLOUT) && IsConnectionSafe(sock))
{
if (_connections[sock]->_send_cb)
_connections[sock]->_send_cb(_connections[sock]);
}
}
}
//服务器的主循环，不断调用 Dispatcher 处理事件。
void Loop()
{
_quit = false; //false服务器开始运行
// AddConnection();
while (!_quit)
{
// Dispatcher(3000);
Dispatcher(-1); //等待epoll事件发生 -1表示无限等待
PrintConnection();
}
_quit = true; //结束
}
//打印所有连接的套接字描述符和输入缓冲区内容。
void PrintConnection()
{
std::cout << "_connections fd list: ";
for (auto &connection : _connections)
{
std::cout << connection.second->SockFd() << ", ";
std::cout << "inbuffer: " << connection.second->Inbuffer().c_str();
}
std::cout << std::endl;
}
~TcpServer()
{
}
private:
std::shared_ptr<Epoller> _epoller_ptr; // 内核
std::shared_ptr<Sock> _listensock_ptr; // 监听socket，可以把他移除到外部
std::unordered_map<int, std::shared_ptr<Connection>> _connections;//存储所有连接的映射，键位套接字描述符，值位connection对象的智能指针
struct epoll_event revs[num];
uint16_t _port;
bool _quit;
// 让上层处理信息
func_t _OnMessage; //一个函数对象用于处理接收到的消息，
};

复制代码

#pragma once
#include <iostream>
#include <cstring>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <fcntl.h>
#include "Log.hpp"
//定义四个枚举类型表示不同的错误类型
enum
{
SocketErr = 2,
BindErr,
ListenErr,
NON_BOLCK_ERR
};
class Sock
{
const static int backlog = 20; //等待队列的最大长度
public:
Sock()
{
}
~Sock()
{
}
public:
void Socket()
{
//调用socket函数创建一个基于IPv4（AF_INET）的TCP（SOCK_STREAM）套接字，并将返回的套接字描述符春初在成员变量sockfd_中
sockfd_ = socket(AF_INET, SOCK_STREAM, 0);
if (sockfd_ < 0)
{
lg(Fatal, "socker error, %s: %d", strerror(errno), errno);
exit(SocketErr);
}
int opt = 1;
//调用 setsockopt 函数设置套接字选项，允许地址和端口的重用
setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));
}
//将套接字绑定到指定的端口
void Bind(uint16_t port)
{
struct sockaddr_in local; //定义一个sockaddr_in结构体变量local，用于存储本地地址信息
memset(&local, 0, sizeof(local));//使用 memset 函数将 local 结构体的内容清零。
local.sin_family = AF_INET; //设置地址族为IPv4
local.sin_port = htons(port);//端口号从主机字节序转化为网络字节序，并存储在local.sin_port中
local.sin_addr.s_addr = INADDR_ANY;//设置 IP 地址为任意地址
//调用bind将socket绑定到指定的地址和端口
if (bind(sockfd_, (struct sockaddr *)&local, sizeof(local)) < 0)
{
lg(Fatal, "bind error, %s: %d", strerror(errno), errno);
exit(BindErr);
}
}
//将socket设置为监听状态
void Listen()
{
//调用listen函数，将socket设置为监听状态，backlog 是等待连接队列的最大长度
if (listen(sockfd_, backlog) < 0)
{
lg(Fatal, "listen error, %s: %d", strerror(errno), errno);
exit(ListenErr);
}
}
//接受客户端的连接请求
int Accept(std::string *clientip, uint16_t *clientport)
{
struct sockaddr_in peer; //定义一个sockaddr_in 结构体变量peer，用于存储客户端的地址信息
socklen_t len = sizeof(peer);//定义一个len表示peer的结构体长度
//调用accept函数接受客户端连接请求，返回新的套接字描述符
int newfd = accept(sockfd_, (struct sockaddr *)&peer, &len);
if (newfd < 0)
{
lg(Warning, "accept error, %s: %d", strerror(errno), errno);
return -1;
}
char ipstr[64];//存储客户端的ip地址字符串
inet_ntop(AF_INET, &peer.sin_addr, ipstr, sizeof(ipstr));//将客户端的ip地址从二进制转化为字符串形式
*clientip = ipstr;//将客户端的IP地址存储到clientip指向的字符串中
*clientport = ntohs(peer.sin_port);//将客户端的端口号从网络字节序转换为主机字节序，并存储到 clientport 指向的变量中。
return newfd;
}
//用于连接到指定的ip地址和端口
bool Connect(const std::string &ip, const uint16_t &port)
{
struct sockaddr_in peer;//存储目标地址信息
memset(&peer, 0, sizeof(peer));//清零
peer.sin_family = AF_INET;//设置地址族为 IPv4。
peer.sin_port = htons(port);//将端口号从主机字节序转换为网络字节序。
inet_pton(AF_INET, ip.c_str(), &(peer.sin_addr));//调用 inet_pton 函数将 IP 地址从字符串形式转换为二进制形式。
//调用 connect 函数连接到指定的地址和端口，返回值存储在 n 中
int n = connect(sockfd_, (struct sockaddr*)&peer, sizeof(peer));
if(n == -1)
{
std::cerr << "connect to " << ip << ":" << port << " error" << std::endl;
return false;
}
return true;
}
void Close()
{
close(sockfd_);
}
int Fd()
{
return sockfd_;
}
private:
int sockfd_;
};

复制代码