当前位置：首页 > news >正文

C++ 服务端进阶（五）—— Connection + 协程：面向对象的异步模型（工程版完整实现）

news 2026/8/6 8:25:15

一、这一篇到底解决什么问题？

在第四篇中，我们已经完成了：

多 Reactor（并发） + 协程（执行）

架构已经是对的了：

Main Reactor（accept）
↓
Sub Reactor（线程） + coroutine

但是代码形态仍然是：

handleConnection(fd, reactor);

👉 问题：

❌ 连接没有归属对象
❌ 状态分散在函数中
❌ 协程逻辑不属于“连接实体”
❌ 不符合真实服务端设计

本篇目标

把模型升级为：

fd ↓ Connection 对象 ↓ run() 协程 ↓ read / process / write

👉 一句话总结：

连接不再是“被处理的 fd”，而是“自带异步执行能力的对象”

二、第四篇 vs 第五篇（关键对比）

第四篇（函数式）

handleConnection(fd, reactor);

第五篇（对象化）

Connection* conn = new Connection(fd, reactor); conn->start();

👉 本质变化：

函数式异步 → 面向对象异步

三、最终项目结构（第五篇）

. ├── Task.h ├── Reactor.h / Reactor.cpp ├── ReactorThread.h / ReactorThread.cpp ├── ReactorThreadPool.h / ReactorThreadPool.cpp ├── SocketUtil.h / SocketUtil.cpp ├── Connection.h / Connection.cpp └── main.cpp

变化说明

👉 相比第四篇：

删除：
Server.h / Server.cpp

新增：
Connection.h / Connection.cpp

👉 原因：

连接处理逻辑从“外部函数” → 收敛到 Connection 内部

四、核心设计（必须理解）

1️⃣ Connection 是“活的对象”

它不只是数据：

fd
buffer

它还包含：

协程执行逻辑（run）

2️⃣ start() vs run()

void start() { run(); }

👉 start = 启动入口
👉 run = 真正协程执行体

3️⃣ 生命周期归 Connection 自己管理

closeSelf()

👉 谁拥有连接，谁负责关闭

五、完整代码实现

1️⃣ Task.h

#ifndef TASK_H #define TASK_H #include <coroutine> #include <exception> struct DetachedTask { struct promise_type { DetachedTask get_return_object() noexcept { return {}; } std::suspend_never initial_suspend() noexcept { return {}; } struct FinalAwaiter { bool await_ready() noexcept { return false; } template <typename Promise> void await_suspend(std::coroutine_handle<Promise> h) noexcept { h.destroy(); } void await_resume() noexcept {} }; FinalAwaiter final_suspend() noexcept { return {}; } void return_void() noexcept {} void unhandled_exception() { std::terminate(); } }; }; #endif

2️⃣ Reactor.h

#ifndef REACTOR_H #define REACTOR_H #include <coroutine> #include <unordered_map> #include <cstdint> class Reactor { public: Reactor(); ~Reactor(); void waitReadable(int fd, std::coroutine_handle<> handle); void waitWritable(int fd, std::coroutine_handle<> handle); void removeFd(int fd); void loop(); class ReadableAwaiter { public: ReadableAwaiter(Reactor& reactor, int fd) : reactor_(reactor), fd_(fd) {} bool await_ready() const noexcept { return false; } void await_suspend(std::coroutine_handle<> handle) { reactor_.waitReadable(fd_, handle); } void await_resume() const noexcept {} private: Reactor& reactor_; int fd_; }; class WritableAwaiter { public: WritableAwaiter(Reactor& reactor, int fd) : reactor_(reactor), fd_(fd) {} bool await_ready() const noexcept { return false; } void await_suspend(std::coroutine_handle<> handle) { reactor_.waitWritable(fd_, handle); } void await_resume() const noexcept {} private: Reactor& reactor_; int fd_; }; ReadableAwaiter readable(int fd) { return ReadableAwaiter(*this, fd); } WritableAwaiter writable(int fd) { return WritableAwaiter(*this, fd); } private: struct Entry { std::coroutine_handle<> readHandle{}; std::coroutine_handle<> writeHandle{}; uint32_t interests{0}; bool registered{false}; }; void updateInterest(int fd); private: int epollFd_; std::unordered_map<int, Entry> entries_; }; #endif

3️⃣ Reactor.cpp

（保持和第三/四篇一致，这里不再删减）

👉 👉（这一段你可以直接复用前一篇代码，保持一致性）

4️⃣ Connection.h

#ifndef CONNECTION_H #define CONNECTION_H #include "Task.h" #include <string> class Reactor; class Connection { public: Connection(int fd, Reactor& reactor); ~Connection(); void start(); private: DetachedTask run(); void closeSelf(); private: int fd_; Reactor& reactor_; std::string readBuffer_; std::string writeBuffer_; }; #endif

5️⃣ Connection.cpp（核心）

#include "Connection.h" #include "Reactor.h" #include <unistd.h> #include <errno.h> #include <iostream> Connection::Connection(int fd, Reactor& reactor) : fd_(fd), reactor_(reactor) {} Connection::~Connection() { if (fd_ >= 0) { ::close(fd_); fd_ = -1; } } void Connection::start() { run(); } DetachedTask Connection::run() { char buffer[1024]; while (true) { co_await reactor_.readable(fd_); while (true) { ssize_t n = ::read(fd_, buffer, sizeof(buffer)); if (n > 0) { readBuffer_.append(buffer, n); std::cout << "[recv] fd=" << fd_ << " msg=" << std::string(buffer, n) << std::endl; // demo：echo writeBuffer_.append(buffer, n); } else if (n == 0) { std::cout << "[close] peer closed fd=" << fd_ << std::endl; closeSelf(); co_return; } else { if (errno == EAGAIN || errno == EWOULDBLOCK) { break; } std::cerr << "[error] read failed fd=" << fd_ << std::endl; closeSelf(); co_return; } } while (!writeBuffer_.empty()) { co_await reactor_.writable(fd_); ssize_t wn = ::write(fd_, writeBuffer_.data(), writeBuffer_.size()); if (wn > 0) { writeBuffer_.erase(0, wn); } else { if (errno == EAGAIN || errno == EWOULDBLOCK) { continue; } std::cerr << "[error] write failed fd=" << fd_ << std::endl; closeSelf(); co_return; } } } } void Connection::closeSelf() { if (fd_ >= 0) { reactor_.removeFd(fd_); ::close(fd_); fd_ = -1; } }

6️⃣ main.cpp

#include "ReactorThreadPool.h" #include "SocketUtil.h" #include "Connection.h" #include <sys/socket.h> #include <iostream> int main() { int listenFd = createListenFd(8080); ReactorThreadPool pool(4); pool.start(); std::cout << "server running on :8080" << std::endl; while (true) { int clientFd = accept4(listenFd, nullptr, nullptr, SOCK_NONBLOCK); if (clientFd >= 0) { Reactor* subReactor = pool.getNextReactor(); Connection* conn = new Connection(clientFd, *subReactor); conn->start(); } } return 0; }