网络编程（4）——异步服务器设计
2025-04-30 12:24:32 # C++ # 网络编程 #异步服务器

五、day5

今天通过昨天学习的异步读写api写一个异步echo服务器

参考：

恋恋风辰官方博客

visual studio配置C++ boost库_哔哩哔哩_bilibili

1）Session类

Session类主要是处理客户端消息收发的会话类，为了简单起见，我们不考虑粘包问题，也不考虑支持手动调用发送的接口，只以应答的方式发送和接收固定长度(1024字节长度)的数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class Session
{
private:
	tcp::socket _socket;
	enum { max_length = 1024 };
	char _data[max_length];

	// headle回调函数
	void headle_read(const boost::system::error_code& error, size_t bytes_transferred);
	void haddle_write(const boost::system::error_code& error);
public:
	Session(boost::asio::io_context& ioc) : _socket(ioc){}
	tcp::socket& Socket() { return _socket; }

	void Start();
};

其中，voidStart()函数通过启动一次异步读操作，准备从客户端读取数据

1
2
3
4
5
6
void Session::Start() {
	memset(_data, 0, max_length); // 缓冲区清零
	// 从套接字中读取数据，并绑定回调函数headle_read
	_socket.async_read_some(boost::asio::buffer(_data, max_length),
		std::bind(&Session::headle_read, this, std::placeholders::_1, std::placeholders::_2));
}

当读取一部分数据后，触发回调函数headle_read()。切记，对boost::asio::async_write、socket.async_read_some、socket.async_send这三个函数的作用和使用场景要充分了解，具体功能我总结在了文章后面。

1
2
3
4
5
6
7
8
9
10
11
void Session::headle_read(const boost::system::error_code& error, size_t bytes_transferred) {
	if (!error) {
		cout << "server receive data is " << _data << endl;
		boost::asio::async_write(_socket, boost::asio::buffer(_data, bytes_transferred),
			std::bind(&Session::haddle_write, this, std::placeholders::_1));
	}
	else {
		cout << "read error" << endl;
		delete this;
	}
}

当读操作没有发生问题时，服务器开始给客户端回传信息，执行async_write()函数，该函数不像async_write_some一样一部分一部分的回传，而是直接一次性的回传我们指定的信息长度。当会传消息长度达到我们指定的长度bytes_transferred时，触发回调函数haddle_write()。

1
2
3
4
5
6
7
8
9
10
11
void Session::haddle_write(const boost::system::error_code& error) {
	if (!error) {
		memset(_data, 0, max_length);
		_socket.async_read_some(boost::asio::buffer(_data, max_length),
			std::bind(&Session::headle_read, this, std::placeholders::_1, std::placeholders::_2));
	}
	else {
		cout << "write error" << error.value() << endl;
		delete this;
	}
}

当写操作没有发生问题时，服务器再一次监听读事件，如果客户端有数据发送过来，那么继续读并触发回调函数headle_read，将读到的消息回传。

这样就达成了一个简单的异步应答服务器的session设计，但是这种服务器并不会在实际生产中使用，因为：

  1. 因为该服务器的发送和接收以应答的方式交互，而并不能做到应用层想随意发送的目的，也就是未做到完全的收发分离(全双工逻辑)。
  2. 该服务器未处理粘包，序列化，以及逻辑和收发线程解耦等问题。
  3. 该服务器存在二次析构的风险。

2）Server类

设计服务器管理接收连接的类：server类

1
2
3
4
5
6
7
8
9
10
11
class Server
{
private:
	void start_accept();  // 启动一个acceptor
	// 当acceptor接收到连接后启动该函数
	void handle_accept(Session* new_session, const boost::system::error_code& error);
	boost::asio::io_context& _ioc;
	tcp::acceptor _acceptor;
public:
	Server(boost::asio::io_context& ioc, short port);
};

start_accept将要接收连接的acceptor绑定到服务上，其内部就是将accpeptor对应的socket描述符绑定到epoll或iocp模型上，实现事件驱动。handle_accept为新连接到来后触发的回调函数。

首先设计Server类的构造函数，用于初始化服务器对象，绑定 I/O 上下文和监听的端口，并启动服务器。

1
2
3
4
5
6
7
// 初始化服务器对象，绑定 I/O 上下文和监听的端口，并启动服务器
Server::Server(boost::asio::io_context& ioc, short port) : _ioc(ioc),
_acceptor(ioc, tcp::endpoint(tcp::v4(), port)) {
	cout << "Server start success, on port: " << port << endl;
	// 开始异步地接受客户端连接请求。服务器启动后就进入等待客户端连接的状态
	start_accept();
}

然后，start_accept()函数启动一个新的异步接受操作，等待客户端连接。但这里有一个问题，为什么所有的session都共用相同的io_context？这个问题我会在后面回答。

1
2
3
4
5
6
7
8
void Server::start_accept() {，
	// 创建一个新的 Session 对象，表示一个与客户端的会话。每个 Session 对象负责处理一个客户端连接。
	// Session 的构造函数接收 _ioc 作为参数，因为 Session 也需要处理异步操作
	Session* new_session = new Session(_ioc);
	// 开始一个异步接受操作，当new_session的socket与客户端连接成功时，调用回调函数handle_accept
	_acceptor.async_accept(new_session->Socket(), std::bind(&Server::handle_accept, this, new_session,
		std::placeholders::_1));
}

服务器检查与客户端的连接是否出错，如果没出错，那么进入session任务中开始重复地读写；反之，删除这个session释放内存；但无论是否出错，服务器都会重新调用start_accept()，保证服务器始终在监听状态，随时准备接收新的连接。

1
2
3
4
5
6
7
8
9
10
// 异步接受操作的回调函数，负责处理客户端连接的结果
void Server::handle_accept(Session* new_session, const boost::system::error_code& error) {
	// 如果没有错误（error 为 false），调用 new_session->Start() 来启动与旧客户端的会话
	if (!error) new_session->Start();
	// 如果发生了错误（如连接失败或出现其他问题），则删除 new_session，释放分配的内存
	else delete new_session;
	// 无论当前连接是否成功，都重新调用 start_accept()，以便服务器能够继续接受下一个新客户端的连接请求。
	// 服务器始终保持在监听状态，随时准备接受新连接
	start_accept();
}

3）客户端

客户端仍使用day2的同步模式代码，因为客户端不需要异步的方式，因为客户端并不是以并发为主，当然后续会继续改进，写为异步收发的方式。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
#include <boost/asio.hpp>
#include <iostream>
using namespace boost::asio::ip;
using std::cout;
using std::endl;
const int MAX_LENGTH = 1024; // 发送和接收的长度为1024字节

int main()
{
    try {
        boost::asio::io_context ioc; // 创建上下文服务
        // 127.0.0.1是本机的回路地址，也就是服务器和客户端在一个机器上
        tcp::endpoint remote_ep(address::from_string("127.0.0.1"), 10086); // 构造endpoint
        tcp::socket sock(ioc);
        boost::system::error_code error = boost::asio::error::host_not_found; // 错误：主机未找到
        sock.connect(remote_ep, error);
        if (error) {
            cout << "connect failed, code is " << error.value() << " error msg is " << error.message() << endl;;
        }
        
        cout << "Enter message: "; // 连接成功，请输入发送的信息
        char request[MAX_LENGTH];
        std::cin.getline(request, MAX_LENGTH);
        size_t request_length = strlen(request);
        boost::asio::write(sock, boost::asio::buffer(request, request_length)); // 一次性发送数据

        char reply[MAX_LENGTH]; // 记录对端回复的信息
        size_t reply_length = boost::asio::read(sock, boost::asio::buffer(reply, request_length));
        cout << "Reply is: ";
        cout.write(reply, reply_length);
        cout << "\n";

    }
    catch (std::exception& e) {
        std::cerr << "Exception: " << e.what() << endl;
    }
    return 0;
}

4）主函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
#include "Session.h"

int main()
{
    try {
        boost::asio::io_context ioc;
        Server s(ioc, 10086);
        ioc.run();
    }
    catch (std::exception& e) {
        std::cerr << "Exception: " << e.what() << '\n';
    }

    return 0;
}

1.为什么服务器读操作使用async_read_some()而不是async_receive()，写操作使用async_write()而不是async_write_some()？

1）为什么读使用 async_read_some

在读取数据时，服务器不知道客户端会发送多少数据。特别是在处理流式数据（如网络请求、持续通信）的情况下，服务器并不一定会一次性接收完所有数据。async_read_some 能够立即处理部分到达的数据，这样在需要时可以继续读取，避免阻塞。使用 async_read_some 读取部分数据后，可以根据当前接收到的数据决定是否需要继续读取更多数据。

2）为什么写使用 async_write

写入完整性要求高：在发送数据时，通常希望将完整的消息一次性发送到对方。如果只写入部分数据，可能会导致对方接收到不完整的数据包，这样可能会破坏协议的完整性。async_write 保证数据完全写入：它确保给定的缓冲区数据会全部写入。如果数据较大，async_write 会处理数据的分段写入，并且会自动继续发送，直到所有数据都写完为止。这样，用户无需自己管理每次写入的进度。

1
2
boost::asio::async_write(_socket, boost::asio::buffer(_data, bytes_transferred),
    std::bind(&Session::haddle_write, this, std::placeholders::_1));

这里 async_write 会确保 _data 中的所有数据都被发送，直到整个数据缓冲区被写入远程端。如果使用 async_write_some，那么程序需要手动处理“剩余数据”的写入，增加了复杂性。

3）总结

读使用 async_read_some：是为了能够处理未知长度的数据流，尤其是当不能确定数据会一次性到达时，允许部分读取并决定是否继续读取。

写使用 async_write：是为了确保将完整的数据写入对方，不需要开发者自己处理部分写入的复杂性。async_write 内部会自动处理多次写入，直到数据完全发送完毕。

2.boost::asio::async_write?socket.async_read_some?socket.async_send？有什么区别

函数 特性 适用场景
async_write 保证缓冲区中数据全部传输，自动管理分批写入 TCP连接、大数据块传输、需要保证完整性
async_read_some 只读取部分数据，不保证读取到完整的数据 流式数据读取、数据长度不确定的情况
async_send 不保证数据的完整性，适用于数据报传输（如UDP） 无连接通信、数据包传输、轻量级传输

解释：

1）boost::asio::async_write()

  • 它会自动管理数据的分段写入。即使一次不能将所有数据写入，async_write 也会继续写入直到缓冲区的数据完全发送给远程端。
  • 开发者不需要担心数据的部分写入情况，async_write 会确保数据的完整性。
  • 适用于传输完整的数据消息或需要确保一次性传输全部内容的场景，如发送完整的HTTP响应或固定长度的数据包。
1
2
boost::asio::async_write(socket, boost::asio::buffer(data), 
  std::bind(&write_handler, std::placeholders::_1, std::placeholders::_2));

在该例子中，async_write 会将 data 中的数据全部发送完毕，才会调用 write_handler 回调。

2）socket.async_read_some()

  • 它的作用是尽快读取数据，即使只读取到一部分数据也会返回。这种行为特别适用于数据流的处理，适合在不知道具体数据长度的情况下使用。
  • async_read_some 并不会等待所有数据到齐后再返回，而是读取到部分数据后立刻调用回调函数处理已到达的数据。
  • 通常与不确定的数据流一起使用，例如服务器从客户端读取未知长度的请求数据时。
1
2
socket.async_read_some(boost::asio::buffer(buffer),
  std::bind(&read_handler, std::placeholders::_1, std::placeholders::_2));

在该例子中，async_read_some 尽量读取一些数据，read_handler 回调处理接收到的数据。

3）socket.async_send()

  • 适用于面向数据报的通信（如 UDP），它发送的数据可能会被分割或丢失，因此不保证数据的可靠性和顺序性。
  • 仅发送一部分数据（即使是一次调用），并且不像 async_write 那样保证缓冲区的数据全部传输。
  • 使用在数据包传输的场景下，可以快速发送数据，而不需要等待确认全部数据被对方接收。
1
2
socket.async_send(boost::asio::buffer(data), 
  std::bind(&send_handler, std::placeholders::_1, std::placeholders::_2));

在该例子中，async_send 将尽可能快地发送 data 数据包，send_handler 回调处理发送结果。

3. 在Server类中，为什么所有的session都共用相同的io_context？

主要原因是Boost.Asio 的设计思想是基于 I/O 上下文（io_context） 来管理异步操作的。

1）统一的异步事件管理

io_context 负责管理所有异步操作的执行。如果每个 Session 都有自己的 io_context，那么每个会话都会有自己独立的事件循环和任务队列，导致以下问题：

效率低下：每个 Session 独立管理自己的异步任务会引入额外的开销，特别是在高并发环境中，这样的设计会浪费大量系统资源（如线程和 CPU 时间片）。

不易管理：通过单个 io_context，所有的异步任务由同一个事件循环调度，统一管理更容易。开发者只需要调用一次 io_context.run()，即可处理所有的异步操作。

2）I/O 多路复用

io_context 支持将多个 I/O 任务放在同一个事件循环中进行管理，这样可以最大化利用操作系统的 I/O 多路复用机制（如 epoll 在 Linux 上）。这样，多个 Session 可以在同一个 io_context 中处理其 I/O 操作，节省系统资源，减少上下文切换。

3）提升并发性

当多个 Session 共用相同的 io_context 时，Boost.Asio 能够利用单个或多个线程来处理所有会话的异步操作。通常情况下，服务器会将 io_context 与多个线程绑定，这样可以提高服务器的并发处理能力。例如：

使用单个 io_context 和多个线程（即 io_context.run() 在多个线程中运行）时，所有线程共享同一个 io_context，可以同时处理不同 Session 中的 I/O 操作，从而提高并发性。

4）简化资源管理

使用单个 io_context 可以简化服务器的资源管理。所有 Session 共用同一个 io_context 后，异步操作完成时，io_context 会自动调度这些回调函数，开发者不需要担心每个 Session 如何分别管理其事件循环。

上一页
2025-04-30 12:24:32 # C++ # 网络编程 #异步服务器
下一页