【项目篇】从零手写高并发服务器（八）：Connection、Acceptor与TcpServer——第一个可运行的服务器

文章目录

• • 从零手写高并发服务器（八）：Connection、Acceptor与TcpServer——第一个可运行的服务器
  • 一、先把Any类加进来
  • 二、Connection模块
  • • 2.1 Connection的作用
    • 2.2 连接状态
    • 2.3 Connection类实现
  • 三、Acceptor模块
  • • 3.1 Acceptor的作用
    • 3.2 Acceptor类实现
  • 四、TcpServer模块
  • • 4.1 TcpServer的作用
    • 4.2 TcpServer类实现
  • 五、第一个EchoServer测试
  • 六、提交代码
  • 七、本篇总结

从零手写高并发服务器（八）：Connection、Acceptor与TcpServer——第一个可运行的服务器

💬 开篇：这是最激动人心的一篇！我们要把之前所有的模块组装起来，实现Connection（连接管理）、Acceptor（监听管理）和TcpServer（服务器整体封装），最终跑起来第一个完整的Echo服务器。

👍 点赞、收藏与分享：写完这篇，你的服务器就真正能跑了！

🚀 循序渐进：Any类 → Connection → Acceptor → TcpServer → EchoServer。

一、先把Any类加进来

Connection模块需要保存协议上下文，用Any类型。我们在前置知识篇已经实现过了，直接加到 server.hpp 中（放在Buffer之后）：

// ==================== Any通用类型 ====================
class Any {
public:
Any() : _content(NULL) {}
template<typename T>
Any(const T &val) : _content(new holder<T>(val)) {}
Any(const Any &other) : _content(other._content ? other._content->clone() : NULL) {}
~Any() { if (_content) delete _content; }

Any &swap(Any &other) {
std::swap(_content, other._content);
return *this;
}

template<typename T>
T *get() {
assert(typeid(T) == _content->type());
return &((holder<T>*)_content)->val;
}

template<typename T>
Any &operator=(const T &val) {
Any(val).swap(*this);
return *this;
}

Any &operator=(Any other) {
other.swap(*this);
return *this;
}

private:
class placeholder {
public:
virtual ~placeholder() {}
virtual const std::type_info &type() = 0;
virtual placeholder *clone() = 0;
};
template<typename T>
class holder : public placeholder {
public:
holder(const T &v) : val(v) {}
const std::type_info &type() { return typeid(T); }
placeholder *clone() { return new holder(val); }
public:
T val;
};
placeholder *_content;
};

二、Connection模块

2.1 Connection的作用

Connection是对一个通信连接的完整管理，每一个通过accept获取到的新连接都会用一个Connection对象来管理。

Connection包含：

┌─────────────────────────────────────────┐
│ Connection │
│ │
│ Socket _socket → 系统IO操作 │
│ Channel _channel → 事件管理 │
│ Buffer _in_buffer → 接收缓冲区 │
│ Buffer _out_buffer → 发送缓冲区 │
│ Any _context → 协议上下文 │
│ EventLoop *_loop → 所属事件循环 │
│ │
│ 回调函数： │
│ _connected_callback → 连接建立完成 │
│ _message_callback → 新数据到来 │
│ _closed_callback → 连接关闭 │
│ _event_callback → 任意事件 │
│ _server_closed_callback → 服务器内部关闭 │
│ │
│ 对外接口： │
│ Send() → 发送数据 │
│ Shutdown() → 关闭连接 │
└─────────────────────────────────────────┘

2.2 连接状态

// 连接状态
typedef enum {
DISCONNECTED, // 连接关闭状态
CONNECTING, // 连接建立成功-待处理状态
CONNECTED, // 连接建立完成，可以通信的状态
DISCONNECTING // 待关闭状态
} ConnStatu;

2.3 Connection类实现

// ==================== Connection连接管理模块 ====================
class Connection;
using PtrConnection = std::shared_ptr<Connection>;

class Connection : public std::enable_shared_from_this<Connection> {
private:
uint64_t _conn_id;
int _sockfd;
bool _enable_inactive_release;
EventLoop *_loop;
ConnStatu _statu;
Socket _socket;
Channel _channel;
Buffer _in_buffer;
Buffer _out_buffer;
Any _context;

using ConnectedCallback = std::function<void(const PtrConnection &)>;
using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>;
using ClosedCallback = std::function<void(const PtrConnection &)>;
using AnyEventCallback = std::function<void(const PtrConnection &)>;
ConnectedCallback _connected_callback;
MessageCallback _message_callback;
ClosedCallback _closed_callback;
AnyEventCallback _event_callback;
ClosedCallback _server_closed_callback;

private:
// ET 模式必须循环读取直到 EAGAIN
void HandleRead() {
while (true) {
char buf[65536];
ssize_t ret = _socket.NonBlockRecv(buf, 65535);
if (ret < 0) {
// EAGAIN 表示没数据了，退出循环
break;
}
if (ret == 0) {
return Release();
}
_in_buffer.WriteAndPush(buf, ret);
}
if (_in_buffer.ReadAbleSize() > 0) {
return _message_callback(shared_from_this(), &_in_buffer);
}
}

// ET 模式必须循环发送直到缓冲区满
void HandleWrite() {
while (_out_buffer.ReadAbleSize() > 0) {
ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(), _out_buffer.ReadAbleSize());
if (ret < 0) {
// 发送缓冲区满了，退出等下次通知
return;
}
_out_buffer.MoveReadOffset(ret);
}
if (_out_buffer.ReadAbleSize() == 0) {
_channel.DisableWrite();
if (_statu == DISCONNECTING) Release();
}
}

void HandleClose() {
if (_in_buffer.ReadAbleSize() > 0) _message_callback(shared_from_this(), &_in_buffer);
return Release();
}
void HandleError() { return HandleClose(); }
void HandleEvent() {
if (_enable_inactive_release) _loop->TimerRefresh(_conn_id);
if (_event_callback) _event_callback(shared_from_this());
}
void EstablishedInLoop() {
assert(_statu == CONNECTING);
_statu = CONNECTED;
_channel.Tie(shared_from_this());
_channel.EnableRead();
if (_connected_callback) _connected_callback(shared_from_this());
}
void ReleaseInLoop() {
if (_statu == DISCONNECTED) return;
_statu = DISCONNECTED;
_channel.Remove();
_socket.Close();
if (_loop->HasTimer(_conn_id)) CancelInactiveReleaseInLoop();
if (_closed_callback) _closed_callback(shared_from_this());
if (_server_closed_callback) _server_closed_callback(shared_from_this());
}
void SendInLoop(Buffer buf) {
if (_statu == DISCONNECTED) return;
_out_buffer.WriteBufferAndPush(buf);
if (!_channel.WriteAble()) _channel.EnableWrite();
}
void ShutdownInLoop() {
_statu = DISCONNECTING;
if (_in_buffer.ReadAbleSize() > 0 && _message_callback) _message_callback(shared_from_this(), &_in_buffer);
if (_out_buffer.ReadAbleSize() == 0) Release();
else if (!_channel.WriteAble()) _channel.EnableWrite();
}
void EnableInactiveReleaseInLoop(int sec) {
_enable_inactive_release = true;
if (_loop->HasTimer(_conn_id)) return _loop->TimerRefresh(_conn_id);
_loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));
}
void CancelInactiveReleaseInLoop() {
_enable_inactive_release = false;
if (_loop->HasTimer(_conn_id)) _loop->TimerCancel(_conn_id);
}

public:
Connection(EventLoop *loop, uint64_t conn_id, int sockfd)
: _conn_id(conn_id), _sockfd(sockfd), _enable_inactive_release(false), _loop(loop),
_statu(CONNECTING), _socket(_sockfd), _channel(loop, _sockfd)
{
_channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
_channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
_channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
_channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
_channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
}
~Connection() { DBG_LOG("RELEASE CONNECTION:%p", this); }
int Fd() { return _sockfd; }
int Id() { return _conn_id; }
bool Connected() { return (_statu == CONNECTED); }
void SetContext(const Any &context) { _context = context; }
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) { _event_callback = cb; }
void SetSrvClosedCallback(const ClosedCallback &cb) { _server_closed_callback = cb; }
void Established() { _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this)); }
void Send(const char *data, size_t len) {
Buffer buf; buf.WriteAndPush(data, len);
_loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
}
void Shutdown() { _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this)); }
void Release() { _loop->RunInLoop(std::bind(&Connection::ReleaseInLoop, this)); }
void EnableInactiveRelease(int sec) { _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec)); }
void CancelInactiveRelease() { _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this)); }
void Upgrade(const Any &context, const ConnectedCallback &conn, const MessageCallback &msg, const ClosedCallback &closed, const AnyEventCallback &event) {
_loop->AssertInLoop();
_context = context; _connected_callback = conn; _message_callback = msg; _closed_callback = closed; _event_callback = event;
}
};

我们还需要在EventLoop中补一个 AssertInLoop 方法：

// 在EventLoop类中添加
void AssertInLoop() {
assert(_thread_id == std::this_thread::get_id());
}

三、Acceptor模块

3.1 Acceptor的作用

Acceptor管理监听套接字，职责很简单：

3.2 Acceptor类实现

// ==================== Acceptor监听管理模块 ====================
class Acceptor {
private:
Socket _socket; // 用于创建监听套接字
EventLoop *_loop; // 用于对监听套接字进行事件监控
Channel _channel; // 用于对监听套接字进行事件管理

using AcceptCallback = std::function<void(int)>;
AcceptCallback _accept_callback;

private:
// 监听套接字的读事件回调–获取新连接
void HandleRead() {
int newfd = _socket.Accept();
if (newfd < 0) {
return;
}
if (_accept_callback) _accept_callback(newfd);
}

int CreateServer(int port) {
bool ret = _socket.CreateServer(port);
assert(ret == true);
return _socket.Fd();
}

public:
Acceptor(EventLoop *loop, int port)
: _socket(CreateServer(port)), _loop(loop), _channel(loop, _socket.Fd())
{
_channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
}

void SetAcceptCallback(const AcceptCallback &cb) { _accept_callback = cb; }
void Listen() { _channel.EnableRead(); }
};

四、TcpServer模块

4.1 TcpServer的作用

TcpServer是整个服务器框架的最终封装，把所有模块串起来：

TcpServer的职责：

┌──────────────────────────────────────────┐
│ TcpServer │
│ │
│ Acceptor → 监听新连接 │
│ LoopThreadPool → 管理子线程EventLoop │
│ EventLoop → 主线程事件循环(baseloop) │
│ connections → 管理所有Connection │
│ │
│ 流程： │
│ 1. Acceptor获取新连接fd │
│ 2. 从线程池中轮询选一个EventLoop │
│ 3. 创建Connection对象，绑定到该EventLoop │
│ 4. Connection自己管理后续的读写事件 │
└──────────────────────────────────────────┘

4.2 TcpServer类实现

// ==================== TcpServer服务器模块 ====================
class TcpServer {
private:
uint64_t _next_id;
uint64_t _timer_id;
int _port;
int _timeout;
bool _enable_inactive_release;
EventLoop _baseloop;
Acceptor _acceptor;
LoopThreadPool _pool;
std::unordered_map<uint64_t, PtrConnection> _conns;

using ConnectedCallback = std::function<void(const PtrConnection &)>;
using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>;
using ClosedCallback = std::function<void(const PtrConnection &)>;
using AnyEventCallback = std::function<void(const PtrConnection &)>;
ConnectedCallback _connected_callback;
MessageCallback _message_callback;
ClosedCallback _closed_callback;
AnyEventCallback _event_callback;

private:
void NewConnection(int fd) {
_next_id++;
PtrConnection conn(new Connection(_pool.NextLoop(), _next_id, fd));

// 【修正】直接设置非阻塞，不要创建 Socket(fd) 临时对象！
int flag = fcntl(fd, F_GETFL, 0);
fcntl(fd, F_SETFL, flag | O_NONBLOCK);

conn->SetMessageCallback(_message_callback);
conn->SetClosedCallback(_closed_callback);
conn->SetConnectedCallback(_connected_callback);
conn->SetAnyEventCallback(_event_callback);
conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));

if (_enable_inactive_release) conn->EnableInactiveRelease(_timeout);

// 【修正】先保存到 map，再 Established，防止竞态条件
_conns[_next_id] = conn;
conn->Established();
}

void RemoveConnectionInLoop(const PtrConnection &conn) {
_conns.erase(conn->Id());
}

void RemoveConnection(const PtrConnection &conn) {
_baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
}

public:
TcpServer(int port)
: _port(port), _next_id(0), _timer_id(0), _enable_inactive_release(false),
_acceptor(&_baseloop, port), _pool(&_baseloop)
{
_acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
_acceptor.Listen();
}

void SetThreadCount(int count) { _pool.SetThreadCount(count); }
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) { _event_callback = cb; }
void EnableInactiveRelease(int timeout) { _timeout = timeout; _enable_inactive_release = true; }
void RunAfter(const TaskFunc &task, int delay) {
_baseloop.RunInLoop([this, task, delay](){ _baseloop.TimerAdd(++_timer_id, delay, task); });
}
void Start() { _pool.Create(); _baseloop.Start(); }
};

五、第一个EchoServer测试

终于到了激动人心的时刻！我们用TcpServer写一个Echo服务器：

cd ~/TcpServer/test
vim echo_server.cpp

#include "../source/server.hpp"

int main() {
TcpServer server(8500);
server.SetThreadCount(2);
server.EnableInactiveRelease(10); // 10秒无活动则断开

server.SetConnectedCallback([](const PtrConnection &conn) {
DBG_LOG("新连接建立 connid:%d", conn->Id());
});

server.SetMessageCallback([](const PtrConnection &conn, Buffer *buf) {
std::string msg = buf->ReadAsStringAndPop(buf->ReadAbleSize());
DBG_LOG("收到数据[%d]: %s", conn->Id(), msg.c_str());
conn->Send(msg.c_str(), msg.size());
});

server.SetClosedCallback([](const PtrConnection &conn) {
DBG_LOG("连接关闭 connid:%d", conn->Id());
});

server.Start();
return 0;
}

编译运行：

g++ -std=c++17 echo_server.cpp -o echo_server -lpthread
./echo_server

另开终端用之前的客户端测试，或者直接用 telnet：

telnet 127.0.0.1 8500

输入任意内容，服务器会原样返回。

服务端输出：

wsh@VM-16-2-ubuntu:~/TcpServer/test$ ./echo_server
[0x7f8a1b2c3740 18:00:01 ../source/server.hpp:156] SIGPIPE INIT
[0x7f8a1b2c3740 18:00:05 echo_server.cpp:9] 新连接建立 connid:1
[0x7f8a0c001740 18:00:08 echo_server.cpp:14] 收到数据[1]: hello
[0x7f8a0c001740 18:00:10 echo_server.cpp:14] 收到数据[1]: world
[0x7f8a1b2c3740 18:00:20 echo_server.cpp:18] 连接关闭 connid:1

第一个完整的高并发服务器跑起来了！

六、提交代码

cd ~/TcpServer
git add .
git commit -m "实现Connection、Acceptor、TcpServer，完成EchoServer"
git push

七、本篇总结

当前 server.hpp 结构：

// 日志宏
// Buffer类
// Any类
// NetWork + Socket类
// Channel类
// Poller类
// TimerTask + TimerWheel类
// EventLoop类
// Channel::Update/Remove实现
// TimerWheel对外接口实现
// LoopThread + LoopThreadPool类
// Connection类
// Acceptor类
// TcpServer类

💬 下一篇预告：实现HTTP协议支持！HttpRequest、HttpResponse、HttpContext、HttpServer，让服务器能处理HTTP请求！