Boost C++ 库实战：构建高性能即时通讯服务器

Boost C++ 库实战：构建高性能即时通讯服务器 | 极客日志

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); // 这里不会抛出异常

#include <boost/asio.hpp>
int main() {
    // 创建 io_context 对象
    boost::asio::io_context ioc;
    // 在这里注册各种异步操作...
    // 运行事件循环（阻塞直到所有任务完成）
    ioc.run();
    return 0;
}

// 异步读取示例
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;
        }
    });

#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;
            }
        });
}

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();

#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 请求（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::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());

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_;
};

#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"

#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;
    }
}

[GateServer]
Port=8080
Host=0.0.0.0

[Database]
Host=localhost
Port=3306

#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);

#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;             // 轮询下标
};

#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(); // 等待所有工作线程结束
    }
}

#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();
            }
        });
}

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));
}

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;
    // ...
};

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_DATA_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;
        }
    });
}

#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();
        });
}

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;
    }
}

组件	用途	关键方法/类
`io_context`	事件循环核心	`run()`, `stop()`
`io_context::work`	阻止空闲退出	配合 `io_context` 使用
`tcp::acceptor`	监听连接	`async_accept()`
`tcp::socket`	TCP 通信	`async_read_some()`, `async_write()`
`signal_set`	信号处理	`async_wait()`
`beast::http`	HTTP 协议	`async_read()`, `async_write()`
`uuid`	唯一 ID 生成	`random_generator()`
`property_tree`	配置解析	`read_ini()`

// 发送前
host_to_network_short()
// 接收后
network_to_host_short()

io_context.stop();   // 先停止接受新请求
pool->Stop();        // 再停止工作线程
server->Shutdown();  // 最后关闭服务

auto self = shared_from_this();
socket.async_read(..., [self](auto ec, auto len) { ... });

Boost C++ 库实战：构建高性能即时通讯服务器

前言

1. Boost.Asio 基础知识

1.1 io_context：事件循环的核心

1.2 异步操作模型

1.3 TCP Socket 编程

创建 TCP 服务器

异步写入数据

1.4 信号处理

2. Boost.Beast 基础知识

2.1 HTTP 请求与响应

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

HTTP 请求对象

HTTP 响应对象

2.2 异步 HTTP 服务器

3. Boost 其他常用组件

3.1 UUID 生成

3.2 配置文件解析

3.3 字节序转换

4. 项目实战案例分析

4.1 AsioIOServicePool：IO 线程池

4.2 CServer：TCP 服务器

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

4.4 HttpConnection：HTTP 连接处理

4.5 信号优雅退出

5. 总结与最佳实践

5.1 核心要点回顾

5.2 最佳实践

5.3 进阶学习路径

参考资料

更多推荐文章

相关免费在线工具

Boost C++ 库实战：构建高性能即时通讯服务器

前言

1. Boost.Asio 基础知识

1.1 io_context：事件循环的核心

1.2 异步操作模型

1.3 TCP Socket 编程

创建 TCP 服务器

异步写入数据

1.4 信号处理

2. Boost.Beast 基础知识

2.1 HTTP 请求与响应

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

HTTP 请求对象

HTTP 响应对象

2.2 异步 HTTP 服务器

3. Boost 其他常用组件

3.1 UUID 生成

3.2 配置文件解析

3.3 字节序转换

4. 项目实战案例分析

4.1 AsioIOServicePool：IO 线程池

4.2 CServer：TCP 服务器

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

4.4 HttpConnection：HTTP 连接处理

4.5 信号优雅退出

5. 总结与最佳实践

5.1 核心要点回顾

5.2 最佳实践

5.3 进阶学习路径

参考资料

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具