【Linux编程】TcpServer 类的计划与实现：构建高性能的 TCP 服务器（二） - ToB企服应用市场:ToB评测及商务社交产业平台

TcpServer::TcpServer(int _port) : TcpServer("0.0.0.0", _port) {
}
TcpServer::TcpServer(std::string _host, int _port) : host(_host), port(_port) {
std::cout << "create tcp server start." << std::endl;
socket_fd = socket(AF_INET, SOCK_STREAM, 0);
if (socket_fd == -1) {
std::cout << "socket create error!";
return;
}
int ret = set_epoll_mode(socket_fd, O_NONBLOCK);
if (ret < 0) {
std::cout << "epoll_mode failed:" << ret << std::endl;
close(socket_fd);
return;
}
server_addr.sin_family = AF_INET;
inet_pton(AF_INET, host.c_str(), &server_addr.sin_addr);
server_addr.sin_port = htons(port);
int opt = 1;
setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
isCreate = true;
std::cout << "create tcp server ok." << std::endl;
}

复制代码

TcpServer::~TcpServer() {
}

复制代码

void TcpServer::Start() {
if (running || !isCreate) {
std::cout << "TcpServer start failed!" << "running=" << running << ", port=" << port << std::endl;
return;
}
auto ret = bind(socket_fd, (struct sockaddr *)&server_addr, sizeof(server_addr));
if (ret == -1) {
std::cout << "bind faild:" << ret << std::endl;
close(socket_fd);
return;
}
ret = listen(socket_fd, SOMAXCONN);
if (ret == -1) {
std::cout << "Listen failed:" << ret << std::endl;
close(socket_fd);
return;
}
std::cout << "server open: " << host << ":" << port << std::endl;
epoll_start();
}

复制代码

void TcpServer::Stop() {
this->Close();
}

复制代码

void TcpServer::Close() {
isCreate = false;
running = false;
socket_event.data.fd = socket_fd;
int ret = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, socket_fd, &socket_event);
ret = close(socket_fd);
std::cout << "socket_fd已关闭:" << ret << std::endl;
for (TcpClient *client : clients)
client->Close();
ret = close(epoll_fd);
std::cout << "TcpServer epoll_fd已关闭:" << ret << std::endl;
}

复制代码

int TcpServer::set_epoll_mode(int sock_fd, int mode) {
int flags = fcntl(sock_fd, F_GETFL, 0);
if (flags == -1) {
std::cout << "epoll_mode failed:" << sock_fd << std::endl;
return -1;
}
return fcntl(sock_fd, F_SETFL, flags | mode);
}

复制代码

void TcpServer::client_accept_thread() {
struct epoll_event client_events[1024];
while (running) {
int ret = epoll_wait(epoll_fd, client_events, 1024, -1);
if (ret < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK)
continue;
else {
std::cerr << "epoll_wait failed: " << ret << " : " << errno << " : " << strerror(errno) << std::endl;
break;
}
}
for (int n = 0; n < ret; ++n) {
if (client_events[n].data.fd == socket_fd) {
client_connect();
} else {
struct epoll_event client_event = client_events[n];
auto client = std::find_if(clients.begin(), clients.end(), [&client_event](const TcpClient *_client)
{ return (_client->client_fd == client_event.data.fd); });
if (client == clients.end())
continue;
int ret = (*client)->data_receive(*client);
if (ret == 0) {
clients.erase(client);
delete *client;
}
}
}
}
std::cout << "服务已关闭，不再提供任何服务！" << std::endl;
isDispose = true;
}

复制代码

void TcpServer::client_connect() {
struct sockaddr_in client_addr;
socklen_t addr_len = sizeof(client_addr);
int client_fd = accept(socket_fd, (struct sockaddr *)&client_addr, &addr_len);
std::cout << "accept:" << client_fd << " : " << strerror(errno) << std::endl;
if (client_fd <= 0)
return;
TcpClient *client = new TcpClient;
client->running = true;
client->connected = true;
client->client_fd = client_fd;
client->local_addr = client_addr;
int ret = set_epoll_mode(client_fd, O_NONBLOCK);
if (ret == -1) {
std::cout << "服务器接受客户端-set_epoll_mode failed:" << strerror(errno) << std::endl;
client->Close();
return;
}
client->add_epoll_event(client_fd, epoll_fd, EPOLLIN | EPOLLET);
client->DataReceived = [this](void *sender, DataReceiveEventArgs *e)
{
DataReceived.Invoke(this, e);
};
client->recv_data = new char[client->recv_data_length];
clients.push_back(client);
std::cout << "新的客户端已接入:" << inet_ntoa(client_addr.sin_addr) << ":" << htons(client_addr.sin_port) << std::endl;
}

复制代码

void TcpServer::epoll_start() {
epoll_fd = epoll_create1(0);
if (epoll_fd == -1) {
std::cout << "poll_create1 failed:" << epoll_fd << std::endl;
close(socket_fd);
return;
}
socket_event.events = EPOLLIN;
socket_event.data.fd = socket_fd;
int ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, socket_fd, &socket_event);
if (ret == -1) {
std::cout << "epoll_ctl failed:" << ret << std::endl;
close(epoll_fd);
return;
}
running = true;
std::thread th = std::thread(&TcpServer::client_accept_thread, this);
th.detach();
}

复制代码

#pragma once
#include <iostream>
#include <vector>
#include <list>
#include <string>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <sys/fcntl.h>
#include <arpa/inet.h>
#include <algorithm>
#include <unistd.h>
#include <thread>
#include "DataReceiveEventArgs.h"
#include "TcpClient.h"
class TcpServer
{
public:
TcpServer(int _port);
TcpServer(std::string _host, int _port);
~TcpServer();
public:
EventHandler<DataReceiveEventArgs> DataReceived;
public:
void Start();
void Stop();
void Close();
bool IsDispose();
private:
void epoll_start(); // epoll初始化（创建epoll）
int set_epoll_mode(int sock_fd, int mode); // epoll模式--创建socket时，为非阻塞模式
void client_accept_thread();
void client_connect();
sockaddr_in get_remote_addr(int sock);
private:
int socket_fd;
int epoll_fd;
std::string host = "0.0.0.0";
int port = 0;
bool running = false;
bool isCreate = false;
int send_buff_size = 1024 * 1024;
int recv_buff_size = 1024 * 1024;
struct sockaddr_in server_addr;
std::list<TcpClient *> clients;
// 将监听套接字添加到 epoll 中，监控 EPOLLIN （表示有数据可读）事件
struct epoll_event socket_event;
bool isDispose = false;
};

复制代码

#include "TcpServer.h"
TcpServer::TcpServer(int _port) : TcpServer("0.0.0.0", _port)
{
}
TcpServer::TcpServer(std::string _host, int _port) : host(_host), port(_port)
{
std::cout << "create tcp server start." << std::endl;
// AF_INET 表示使用 IPv4 协议
// SOCK_STREAM 表示套接字的类型，表示面向连接的流式套接字
socket_fd = socket(AF_INET, SOCK_STREAM, 0);
// std::cout << "create tcp server socket_fd:" << socket_fd << std::endl;
if (socket_fd == -1)
{
std::cout << "socket ceate error!";
return;
}
// 文件描述符为非阻塞模式
int ret = set_epoll_mode(socket_fd, O_NONBLOCK);
if (ret < 0)
{
std::cout << "epoll_mode failed:" << ret << std::endl;
close(socket_fd);
return;
}
server_addr.sin_family = AF_INET;
// ip字符串转int
inet_pton(AF_INET, host.c_str(), &server_addr.sin_addr);
//server_addr.sin_addr.s_addr = ntohl(server_addr.sin_addr.s_addr);
server_addr.sin_port = htons(port);
// 以下设置表示当调用close关闭客户端时，立即释放端口，不等待
// 在TCP服务端，客户端调用 close(client_fd) 关闭连接后，如果你尝试重新连接时出现端口没有完全释放的情况，通常是由于 TCP 连接的 TIME_WAIT 状态没有及时清理。这是 TCP 协议的正常行为。
// 在TCP连接关闭后，端口会进入 TIME_WAIT 状态。这个状态的目的是确保最后的数据包能够正确到达。如果新的连接尝试在该端口上进行，而该端口仍然处于 TIME_WAIT 状态，就会出现端口被占用的情况。
// TIME_WAIT 状态通常会持续一段时间（默认是4分钟，即240秒），这可以通过操作系统的内核参数来修改。
// struct linger linger_opt;
// linger_opt.l_onoff = 1; // 启用
// linger_opt.l_linger = 1; // 立即关闭
// // 此方法亲测无效
// // setsockopt(socket_fd, SOL_SOCKET, SO_LINGER, &linger_opt, sizeof(linger_opt));
int opt = 1;
setsockopt(socket_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
isCreate = true;
std::cout << "create tcp server ok." << std::endl;
}
TcpServer::~TcpServer()
{
}
void TcpServer::Start()
{
if (running || !isCreate)
{
std::cout << "TcpServer start failed!" << "running=" << running << ", port=" << port << std::endl;
// close(socket_fd);
return;
}
// 绑定套接字到指定端口
auto ret = bind(socket_fd, (struct sockaddr *)&server_addr, sizeof(server_addr));
if (ret == -1)
{
std::cout << "bind faild:" << ret << std::endl;
close(socket_fd);
return;
}
// 开始监听
ret = listen(socket_fd, SOMAXCONN);
if (ret == -1)
{
std::cout << "Listen failed:" << ret << std::endl;
close(socket_fd);
return;
}
std::cout << "server open: " << host << ":" << port << std::endl;
epoll_start();
}
void TcpServer::Stop()
{
this->Close();
}
void TcpServer::Close()
{
isCreate = false;
running = false;
// 1. 删除socket_fd的epoll事件
socket_event.data.fd = socket_fd;
int ret = epoll_ctl(epoll_fd, EPOLL_CTL_DEL, socket_fd, &socket_event);
// 2. 关闭监听套接字
ret = close(socket_fd);
std::cout << "socket_fd已关闭:" << ret << std::endl;
// 3. 从 epoll 中移除所有连接的客户端文件描述符
for (TcpClient *client : clients)
client->Close();
// 4. 关闭 epoll 文件描述符
ret = close(epoll_fd);
std::cout << "TcpServer epoll_fd已关闭:" << ret << std::endl;
}
bool TcpServer::IsDispose()
{
return isDispose;
}
int TcpServer::set_epoll_mode(int sock_fd, int mode)
{
/*
O_NONBLOCK（非阻塞模式）：如果设置了这个标志，表示该套接字（或文件）是非阻塞的，执行读写操作时不会阻塞调用进程或线程。
套接字在没有数据可读或可写时不会让程序等待，而是立即返回。
O_RDWR、O_WRONLY、O_RDONLY（访问模式）：表示套接字的打开方式。
O_APPEND（追加模式）：指示文件或套接字在写操作时会追加数据。
*/
int flags = fcntl(sock_fd, F_GETFL, 0); // 获取当前套接字的文件状态标志
if (flags == -1)
{
std::cout << "epoll_mode failed:" << sock_fd << std::endl;
return -1;
}
// 设置套接字为非阻塞模式
return fcntl(sock_fd, F_SETFL, flags | mode);
}
void TcpServer::client_accept_thread()
{
struct epoll_event client_events[1024];
while (running)
{
// 阻塞等待事件
int ret = epoll_wait(epoll_fd, client_events, 1024, -1);
if (ret < 0)
{
if (errno == EAGAIN || errno == EWOULDBLOCK)
continue;
else
{
std::cerr << "epoll_wait failed: " << ret << " : " << errno << " : " << strerror(errno) << std::endl;
break;
}
}
// 处理返回的事件
for (int n = 0; n < ret; ++n)
{
if (client_events[n].data.fd == socket_fd)
{
// 如果是监听套接字的事件,有新的客户端连接
client_connect();
}
else
{
struct epoll_event client_event = client_events[n];
auto client = std::find_if(clients.begin(), clients.end(), [&client_event](const TcpClient *_client)
{ return (_client->client_fd == client_event.data.fd); });
if (client == clients.end())
continue;
// 客户端有数据
int ret = (*client)->data_receive(*client);
if (ret == 0)
{
clients.erase(client);
delete *client;
}
}
}
}
std::cout << "服务已关闭，不再提供任何服务！" << std::endl;
isDispose = true;
}
void TcpServer::client_connect()
{
struct sockaddr_in client_addr;
socklen_t addr_len = sizeof(client_addr);
int client_fd = accept(socket_fd, (struct sockaddr *)&client_addr, &addr_len); // 接受连接
std::cout << "accept:" << client_fd << " : " << strerror(errno) << std::endl;
if (client_fd <= 0)
return;
TcpClient *client = new TcpClient;
client->running = true;
client->connected = true;
client->client_fd = client_fd;
client->local_addr = client_addr;
// 设置客户端套接字为非阻塞模式
int ret = set_epoll_mode(client_fd, O_NONBLOCK);
if (ret == -1)
{
std::cout << "服务器接受客户端-set_epoll_mode failed:" << strerror(errno) << std::endl;
client->Close();
return;
}
// client.SetSendBuffSize(send_buff_size);
// client.SetRecvBuffSize(recv_buff_size);
// 将新客户端套接字添加到 epoll 中，监听可读事件
// client->create_epoll();
client->add_epoll_event(client_fd, epoll_fd, EPOLLIN | EPOLLET);
client->DataReceived = [this](void *sender, DataReceiveEventArgs *e)
{
DataReceived.Invoke(this, e);
};
// client->start_receive();
client->recv_data = new char[client->recv_data_length];
clients.push_back(client);
std::cout << "新的客户端已接入:" << inet_ntoa(client_addr.sin_addr) << ":" << htons(client_addr.sin_port) << std::endl;
}
void TcpServer::epoll_start()
{
// 创建epoll
epoll_fd = epoll_create1(0);
if (epoll_fd == -1)
{
std::cout << "poll_create1 failed:" << epoll_fd << std::endl;
close(socket_fd);
return;
}
// 监听可读事件
socket_event.events = EPOLLIN;
// 将监听套接字的文件描述符传给 epoll
socket_event.data.fd = socket_fd;
// 将监听套接字添加到 epoll 中，监控 EPOLLIN 事件（表示有数据可读）
int ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, socket_fd, &socket_event);
if (ret == -1)
{
std::cout << "epoll_ctl failed:" << ret << std::endl;
close(epoll_fd);
return;
}
running = true;
std::thread th = std::thread(&TcpServer::client_accept_thread, this);
th.detach();
}

复制代码