Boost C++ 库实战教程：打造高性能即时通讯服务器

前言

boost.asio

boost.asio 是一个跨平台的 C++ 网络库。支持同步 IO 和异步IO。

同步 IO

boost::asio::io_context io_context;

boost::asio::ip::tcp::socket socket(io_context);

socket.connect(server_endpoint); // 将会抛出异常

boost::system::error_code ec;

socket.connect(server_endpoint, ec); // 这里不会抛出异常

异步 IO

proactor 模型

详解在：https://www.boost.org/doc/libs/1_82_0/doc/html/boost_asio/overview/core/async.html。 其中：

• Asynchronous Operation Processor
• reactor 模型中， 相当于 select、poll、epoll。
• IOCP 中， 相当于 AcceptEx、WSARecv、WSASend等。

基本概念及接口

• io_context

  类似 reactor 对象，或者 iocp 对象。

  • run

• ip::tcp::acceptor

  • async_accept

• ip::tcp::endpoint：端点，某个地址和端口的封装。

• ip::tcp::socket：socket创建时，需要与 io_context 进行绑定。

  • async_connect
  • async_read_some
  • async_write_some

目录

1. Boost.Asio 基础知识

1.1 io_context：事件循环的核心

io_context 是 Boost.Asio 的核心类，它负责管理所有的异步操作和事件分发。

#include <boost/asio.hpp>

int main() {
// 创建 io_context 对象
boost::asio::io_context ioc;

// 在这里注册各种异步操作…

// 运行事件循环（阻塞直到所有任务完成）
ioc.run();

return 0;
}

关键概念：

• io_context 是一个事件循环，调用 run() 后会阻塞并处理已注册的异步事件
• 当没有待处理的事件时，run() 会返回
• 可以通过 stop() 强制停止事件循环

1.2 异步操作模型

Boost.Asio 采用 Proactor 异步模式：

// 异步读取示例
socket.async_read_some(
boost::asio::buffer(data, max_length),
[this](const boost::system::error_code& ec, std::size_t bytes_transferred) {
if (!ec) {
// 读取成功，处理数据
std::cout << "读取了 " << bytes_transferred << " 字节" << std::endl;
} else {
// 处理错误
std::cerr << "读取错误: " << ec.what() << std::endl;
}
}
);

1.3 TCP Socket 编程

创建 TCP 服务器

#include <boost/asio.hpp>
using boost::asio::ip::tcp;

class Server {
public:
Server(boost::asio::io_context& ioc, short port)
: acceptor_(ioc, tcp::endpoint(tcp::v4(), port)) {
StartAccept();
}

private:
void StartAccept() {
// 创建新的 socket
auto socket = std::make_shared<tcp::socket>(acceptor_.get_executor());

// 异步接受连接
acceptor_.async_accept(*socket,
[this, socket](const boost::system::error_code& ec) {
if (!ec) {
// 连接成功，处理新客户端
HandleClient(socket);
}
// 继续接受下一个连接
StartAccept();
});
}

void HandleClient(std::shared_ptr<tcp::socket> socket) {
// 处理客户端连接
}

tcp::acceptor acceptor_;
};

异步写入数据

void Send(const std::string& message) {
// async_write 保证发送完整数据
boost::asio::async_write(socket_,
boost::asio::buffer(message),
[this](const boost::system::error_code& ec, std::size_t /*bytes*/) {
if (ec) {
std::cerr << "发送失败: " << ec.what() << std::endl;
}
});
}

1.4 信号处理

优雅地处理 SIGINT（Ctrl+C）和 SIGTERM 信号：

boost::asio::io_context ioc;
boost::asio::signal_set signals(ioc, SIGINT, SIGTERM);

signals.async_wait([&ioc](const boost::system::error_code& error, int signal_number) {
if (!error) {
std::cout << "收到信号: " << signal_number << "，正在关闭…" << std::endl;
ioc.stop();
}
});

ioc.run();

2. Boost.Beast 基础知识

Boost.Beast 是基于 Boost.Asio 的 HTTP/WebSocket 库，专门用于网络协议处理。

2.1 HTTP 请求与响应

命名空间别名（推荐写法）

#include <boost/beast.hpp>
#include <boost/beast/http.hpp>

namespace beast = boost::beast;
namespace http = beast::http;
namespace net = boost::asio;
using tcp = boost::asio::ip::tcp;

HTTP 请求对象

// 创建 HTTP 请求（dynamic_body 支持动态大小的请求体）
http::request<http::dynamic_body> request;

// 获取请求方法
if (request.method() == http::verb::get) {
// 处理 GET 请求
} else if (request.method() == http::verb::post) {
// 处理 POST 请求
}

// 获取请求路径
std::string target = request.target();

// 获取请求体内容
std::string body = boost::beast::buffers_to_string(request.body().data());

HTTP 响应对象

http::response<http::dynamic_body> response;

// 设置状态码
response.result(http::status::ok); // 200

// 设置响应头
response.set(http::field::server, "MyServer");
response.set(http::field::content_type, "application/json");

// 设置响应体
beast::ostream(response.body()) << "{\\"status\\": \\"success\\"}";

// 设置 Content-Length
response.content_length(response.body().size());

2.2 异步 HTTP 服务器

class HttpConnection : public std::enable_shared_from_this<HttpConnection> {
public:
HttpConnection(boost::asio::io_context& ioc) : socket_(ioc) {}

void Start() {
auto self = shared_from_this();

// 异步读取 HTTP 请求
http::async_read(socket_, buffer_, request_,
[self](beast::error_code ec, std::size_t bytes_transferred) {
if (!ec) {
self->HandleRequest();
}
});
}

private:
void HandleRequest() {
// 构建响应
response_.result(http::status::ok);
response_.set(http::field::content_type, "text/plain");
beast::ostream(response_.body()) << "Hello, World!";

WriteResponse();
}

void WriteResponse() {
auto self = shared_from_this();

response_.content_length(response_.body().size());

// 异步写入响应
http::async_write(socket_, response_,
[self](beast::error_code ec, std::size_t) {
// 关闭发送通道
self->socket_.shutdown(tcp::socket::shutdown_send, ec);
});
}

tcp::socket socket_;
beast::flat_buffer buffer_{8192};
http::request<http::dynamic_body> request_;
http::response<http::dynamic_body> response_;
};

3. Boost 其他常用组件

3.1 UUID 生成

用于生成唯一的会话标识符：

#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>

// 生成随机 UUID
boost::uuids::uuid uuid = boost::uuids::random_generator()();

// 转换为字符串
std::string session_id = boost::uuids::to_string(uuid);
// 输出示例: "550e8400-e29b-41d4-a716-446655440000"

3.2 配置文件解析

使用 boost::property_tree 读取 INI 配置文件：

#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/ini_parser.hpp>
#include <boost/filesystem.hpp>

// 获取当前工作目录
boost::filesystem::path current_path = boost::filesystem::current_path();

// 拼接配置文件路径
boost::filesystem::path config_path = current_path / "config.ini";

// 读取 INI 文件
boost::property_tree::ptree pt;
boost::property_tree::read_ini(config_path.string(), pt);

// 遍历所有 section 和 key-value
for (const auto& section_pair : pt) {
const std::string& section_name = section_pair.first; // 如 "GateServer"
const boost::property_tree::ptree& section_tree = section_pair.second;

for (const auto& key_value_pair : section_tree) {
const std::string& key = key_value_pair.first;
const std::string& value = key_value_pair.second.get_value<std::string>();
std::cout << section_name << "." << key << " = " << value << std::endl;
}
}

配置文件示例 (config.ini):

[GateServer]
Port=8080
Host=0.0.0.0

[Database]
Host=localhost
Port=3306

3.3 字节序转换

网络传输通常使用大端序（网络字节序），而大多数机器使用小端序：

#include <boost/asio/detail/socket_ops.hpp>

// 主机字节序 -> 网络字节序（发送前）
short msg_id = 1001;
short msg_id_network = boost::asio::detail::socket_ops::host_to_network_short(msg_id);

// 网络字节序 -> 主机字节序（接收后）
short msg_id_host = boost::asio::detail::socket_ops::network_to_host_short(msg_id_network);

4. 项目实战案例分析

下面通过实际项目代码，展示 Boost 在真实服务器开发中的应用。

4.1 AsioIOServicePool：IO 线程池

为什么需要 IO 线程池？

• 单线程 io_context 无法充分利用多核 CPU
• 多个 io_context + 多线程可以提高并发处理能力

头文件 AsioIOServicePool.h:

#pragma once
#include <vector>
#include <boost/asio.hpp>
#include "Singleton.h"

// 使用 io_context 连接池提高并发
class AsioIOServicePool : public Singleton<AsioIOServicePool> {
friend class Singleton<AsioIOServicePool>;
public:
using IOService = boost::asio::io_context;
using Work = boost::asio::io_context::work; // 防止 io_context 在无任务时退出
using WorkPtr = std::unique_ptr<Work>;

~AsioIOServicePool();
AsioIOServicePool(const AsioIOServicePool&) = delete;
AsioIOServicePool& operator=(const AsioIOServicePool&) = delete;

// 轮询获取一个 io_context（负载均衡）
boost::asio::io_context& GetIOService();

// 停止所有线程
void Stop();

private:
AsioIOServicePool(std::size_t size = 2); // 默认 2 个线程

std::vector<IOService> _ioServices; // 多个 io_context
std::vector<WorkPtr> _works; // 每个 io_context 配一个 work
std::vector<std::thread> _threads; // 工作线程
std::size_t _nextIOService; // 轮询下标
};

实现文件 AsioIOServicePool.cpp:

#include "AsioIOServicePool.h"
#include <iostream>
using namespace std;

// 构造函数：创建线程池
AsioIOServicePool::AsioIOServicePool(std::size_t size)
: _ioServices(size), _works(size), _nextIOService(0) {

// 给每个 io_context 绑定一个 work
for (std::size_t i = 0; i < size; ++i) {
_works[i] = std::unique_ptr<Work>(new Work(_ioServices[i]));
}

// 为每个 io_context 创建一个线程运行事件循环
for (std::size_t i = 0; i < _ioServices.size(); ++i) {
_threads.emplace_back([this, i]() {
_ioServices[i].run(); // 线程进入事件循环
});
}
}

AsioIOServicePool::~AsioIOServicePool() {
Stop();
std::cout << "AsioIOServicePool destruct" << endl;
}

// 轮询分配 io_context（Round-Robin 负载均衡）
boost::asio::io_context& AsioIOServicePool::GetIOService() {
auto& service = _ioServices[_nextIOService++];
if (_nextIOService == _ioServices.size()) {
_nextIOService = 0;
}
return service;
}

// 关闭线程池
void AsioIOServicePool::Stop() {
for (auto& work : _works) {
work->get_io_context().stop(); // 告诉 io_context 停止
work.reset(); // 删除 work
}

for (auto& t : _threads) {
t.join(); // 等待所有工作线程结束
}
}

核心知识点：

4.2 CServer：TCP 服务器

GateServer 版本的服务器用于处理 HTTP 连接：

#include "CServer.h"
#include <iostream>
#include "HttpConnection.h"
#include "AsioIOServicePool.h"

// 构造函数：初始化 acceptor
CServer::CServer(boost::asio::io_context& ioc, unsigned short& port)
: _ioc(ioc), _acceptor(ioc, tcp::endpoint(tcp::v4(), port)) {
}

// 开始接受连接
void CServer::Start() {
auto self = shared_from_this();

// 从线程池获取一个 io_context（负载均衡）
auto& io_context = AsioIOServicePool::GetInstance()->GetIOService();

// 创建新的 HTTP 连接对象
std::shared_ptr<HttpConnection> new_con = std::make_shared<HttpConnection>(io_context);

// 异步接受连接
_acceptor.async_accept(new_con->GetSocket(), [self, new_con](beast::error_code ec) {
try {
if (ec) {
// 出错则放弃这个连接，继续监听
self->Start();
return;
}

// 处理新连接
new_con->Start();

// 继续监听下一个连接
self->Start();
}
catch (std::exception& exp) {
std::cout << "exception is " << exp.what() << std::endl;
self->Start();
}
});
}

ChatServer 版本（处理长连接）:

CServer::CServer(boost::asio::io_context& io_context, short port)
: _io_context(io_context), _port(port),
_acceptor(io_context, tcp::endpoint(tcp::v4(), port)) {
cout << "Server start success, listen on port : " << _port << endl;
StartAccept();
}

void CServer::HandleAccept(shared_ptr<CSession> new_session,
const boost::system::error_code& error) {
if (!error) {
new_session->Start(); // 开始处理会话
lock_guard<mutex> lock(_mutex);
_sessions.insert(make_pair(new_session->GetSessionId(), new_session));
}
else {
cout << "session accept failed, error is " << error.what() << endl;
}

StartAccept(); // 继续接受新连接
}

void CServer::StartAccept() {
// 从线程池获取 io_context
auto& io_context = AsioIOServicePool::GetInstance()->GetIOService();

// 创建新会话
shared_ptr<CSession> new_session = make_shared<CSession>(io_context, this);

// 异步接受
_acceptor.async_accept(new_session->GetSocket(),
std::bind(&CServer::HandleAccept, this, new_session, placeholders::_1));
}

4.3 CSession：会话管理与异步读写

这是项目中最核心的类，展示了完整的异步 TCP 通信实现：

头文件关键部分:

class CSession : public std::enable_shared_from_this<CSession> {
public:
CSession(boost::asio::io_context& io_context, CServer* server);
~CSession();

tcp::socket& GetSocket();
std::string& GetSessionId();
void Start();
void Send(std::string msg, short msgid);
void Close();

private:
// 异步读取指定长度
void asyncReadFull(std::size_t maxLength,
std::function<void(const boost::system::error_code&, std::size_t)> handler);
void asyncReadLen(std::size_t read_len, std::size_t total_len,
std::function<void(const boost::system::error_code&, std::size_t)> handler);
void HandleWrite(const boost::system::error_code& error,
std::shared_ptr<CSession> shared_self);

tcp::socket _socket;
std::string _session_id;
char _data[MAX_LENGTH];
CServer* _server;
bool _b_close;
std::queue<shared_ptr<SendNode>> _send_que; // 发送队列
std::mutex _send_lock;
// …
};

构造函数：使用 UUID 生成会话 ID:

CSession::CSession(boost::asio::io_context& io_context, CServer* server)
: _socket(io_context), _server(server), _b_close(false), _b_head_parse(false), _user_uid(0) {
// 生成唯一的会话 ID
boost::uuids::uuid a_uuid = boost::uuids::random_generator()();
_session_id = boost::uuids::to_string(a_uuid);
_recv_head_node = make_shared<MsgNode>(HEAD_TOTAL_LEN);
}

异步发送：

void CSession::Send(std::string msg, short msgid) {
std::lock_guard<std::mutex> lock(_send_lock);
int send_que_size = _send_que.size();

// 防止发送队列过长
if (send_que_size >= MAX_SENDQUE) {
std::cout << "session: " << _session_id << " send que fulled" << endl;
return;
}

// 加入发送队列
_send_que.push(make_shared<SendNode>(msg.c_str(), msg.length(), msgid));

// 如果队列之前非空，说明已有发送在进行，直接返回
if (send_que_size > 0) return;

// 开始发送队列头部的消息
auto& msgnode = _send_que.front();
boost::asio::async_write(_socket,
boost::asio::buffer(msgnode->_data, msgnode->_total_len),
std::bind(&CSession::HandleWrite, this, std::placeholders::_1, SharedSelf()));
}

发送回调处理：

void CSession::HandleWrite(const boost::system::error_code& error,
std::shared_ptr<CSession> shared_self) {
try {
if (!error) {
std::lock_guard<std::mutex> lock(_send_lock);
_send_que.pop(); // 移除已发送的消息

// 继续发送队列中的下一条消息
if (!_send_que.empty()) {
auto& msgnode = _send_que.front();
boost::asio::async_write(_socket,
boost::asio::buffer(msgnode->_data, msgnode->_total_len),
std::bind(&CSession::HandleWrite, this, std::placeholders::_1, shared_self));
}
} else {
std::cout << "handle write failed, error is " << error.what() << endl;
Close();
_server->ClearSession(_session_id);
}
}
catch (std::exception& e) {
std::cerr << "Exception code : " << e.what() << endl;
}
}

异步读取（递归模式）：

// 读取完整长度
void CSession::asyncReadFull(std::size_t maxLength,
std::function<void(const boost::system::error_code&, std::size_t)> handler) {
::memset(_data, 0, MAX_LENGTH); // 清空缓冲区
asyncReadLen(0, maxLength, handler);
}

// 递归读取指定字节数
void CSession::asyncReadLen(std::size_t read_len, std::size_t total_len,
std::function<void(const boost::system::error_code&, std::size_t)> handler) {

auto self = shared_from_this(); // 防止对象被提前销毁

_socket.async_read_some(
boost::asio::buffer(_data + read_len, total_len – read_len),
[read_len, total_len, handler, self](const boost::system::error_code& ec,
std::size_t bytesTransfered) {
if (ec) {
handler(ec, read_len + bytesTransfered);
return;
}

// 检查是否读够了
if (read_len + bytesTransfered >= total_len) {
handler(ec, read_len + bytesTransfered);
return;
}

// 还没读够，继续递归读取
self->asyncReadLen(read_len + bytesTransfered, total_len, handler);
});
}

读取消息头部（含字节序转换）：

void CSession::AsyncReadHead(int total_len) {
auto self = shared_from_this();

asyncReadFull(HEAD_TOTAL_LEN, [self, this](const boost::system::error_code& ec,
std::size_t bytes_transfered) {
try {
if (ec) {
std::cout << "handle read failed, error is " << ec.what() << endl;
Close();
_server->ClearSession(_session_id);
return;
}

_recv_head_node->Clear();
memcpy(_recv_head_node->_data, _data, bytes_transfered);

// 解析消息 ID（2 字节）
short msg_id = 0;
memcpy(&msg_id, _recv_head_node->_data, HEAD_ID_LEN);
// 网络字节序 -> 主机字节序
msg_id = boost::asio::detail::socket_ops::network_to_host_short(msg_id);

// 解析消息长度（2 字节）
short msg_len = 0;
memcpy(&msg_len, _recv_head_node->_data + HEAD_ID_LEN, HEAD_DATA_LEN);
msg_len = boost::asio::detail::socket_ops::network_to_host_short(msg_len);

// 验证消息长度合法性
if (msg_len > MAX_LENGTH) {
std::cout << "invalid data length is " << msg_len << endl;
_server->ClearSession(_session_id);
return;
}

// 创建消息节点，继续读取消息体
_recv_msg_node = make_shared<RecvNode>(msg_len, msg_id);
AsyncReadBody(msg_len);
}
catch (std::exception& e) {
std::cout << "Exception code is " << e.what() << endl;
}
});
}

4.4 HttpConnection：HTTP 连接处理

#include "HttpConnection.h"
#include "LogicSystem.h"

HttpConnection::HttpConnection(boost::asio::io_context& ioc)
: _socket(ioc) {
}

// 开启监听
void HttpConnection::Start() {
auto self = shared_from_this();

http::async_read(_socket, _buffer, _request,
[self](beast::error_code ec, std::size_t bytes_transferred) {
try {
if (ec) {
std::cout << "http read err is " << ec.what() << std::endl;
return;
}

boost::ignore_unused(bytes_transferred);
self->HandleReq();
self->CheckDeadline();
}
catch (std::exception& exp) {
std::cout << "exception is " << exp.what() << std::endl;
}
});
}

// 处理 HTTP 请求
void HttpConnection::HandleReq() {
_response.version(_request.version());
_response.keep_alive(false); // 短连接

if (_request.method() == http::verb::get) {
PreParseGetParam();
bool success = LogicSystem::GetInstance()->HandleGet(_get_url, shared_from_this());
if (!success) {
_response.result(http::status::not_found);
_response.set(http::field::content_type, "text/plain");
beast::ostream(_response.body()) << "url not found\\r\\n";
WriteResponse();
return;
}

_response.result(http::status::ok);
_response.set(http::field::server, "GateServer");
WriteResponse();
return;
}

if (_request.method() == http::verb::post) {
// 从请求体获取 JSON 数据
auto body_str = boost::beast::buffers_to_string(_request.body().data());

bool success = LogicSystem::GetInstance()->HandlePost(_request.target(), shared_from_this());
if (!success) {
_response.result(http::status::not_found);
_response.set(http::field::content_type, "text/plain");
beast::ostream(_response.body()) << "url not found\\r\\n";
WriteResponse();
return;
}

_response.result(http::status::ok);
_response.set(http::field::server, "GateServer");
WriteResponse();
}
}

// 超时检测
void HttpConnection::CheckDeadline() {
auto self = shared_from_this();

deadline_.async_wait([self](beast::error_code ec) {
if (!ec) {
// 超时，关闭连接
self->_socket.close(ec);
}
});
}

// 发送响应
void HttpConnection::WriteResponse() {
auto self = shared_from_this();

_response.content_length(_response.body().size());

http::async_write(_socket, _response,
[self](beast::error_code ec, std::size_t) {
// 关闭发送通道
self->_socket.shutdown(tcp::socket::shutdown_send, ec);
self->deadline_.cancel();
});
}

4.5 信号优雅退出

多服务协同关闭的完整示例（ChatServer）：

int main() {
auto& cfg = ConfigMgr::Inst();
auto server_name = cfg["SelfServer"]["Name"];

try {
auto pool = AsioIOServicePool::GetInstance();

// 启动 gRPC 服务器
std::string server_address(cfg["SelfServer"]["Host"] + ":" + cfg["SelfServer"]["RPCPort"]);
ChatServiceImpl service;
grpc::ServerBuilder builder;
builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
builder.RegisterService(&service);
std::unique_ptr<grpc::Server> server(builder.BuildAndStart());

std::thread grpc_server_thread([&server] {
server->Wait(); // gRPC 在独立线程运行
});

// 设置信号处理
boost::asio::io_context io_context;
boost::asio::signal_set signals(io_context, SIGINT, SIGTERM);

signals.async_wait([&io_context, pool, &server](
const boost::system::error_code& error, int signal_number) {
if (error) return;

std::cout << "Received signal: " << signal_number << std::endl;

// 按顺序关闭各个组件
io_context.stop(); // 1. 停止接受新连接
pool->Stop(); // 2. 停止后台 IO 线程
server->Shutdown(); // 3. 最后停止 gRPC
});

// 启动 TCP 服务器
auto port_str = cfg["SelfServer"]["Port"];
CServer s(io_context, atoi(port_str.c_str()));

io_context.run(); // 阻塞运行

// 清理资源
RedisMgr::GetInstance()->Close();
grpc_server_thread.join();
}
catch (std::exception& e) {
std::cerr << "Exception: " << e.what() << endl;
}
}

5. 总结与最佳实践

5.1 核心要点回顾

5.2 最佳实践

5.3 进阶学习路径

参考资料

• Boost.Asio 官方文档
• Boost.Beast 官方文档
• Boost 1.85.0 发布说明