C++网络编程之利用Boost库搭建简单的异步服务器 - ToB企服应用市场:ToB评测及商务社交产业平台

#include "Server.h"
//注意这里的_acceptor的构造方式，接受一个io_context，以及一个端点。这个端点通过一个默认可用的ip和指定端口进行构造。
Server::Server(boost::asio::io_context& ioc, unsigned short port): _ioc(ioc),
_acceptor(ioc, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)) {
std::cout << "Server starts on success, using port: " << port << std::endl;
//启动监听
//这么说不是一个严谨的说法，稍后将看到为什么
start_accept();
}
void Server::start_accept() {
//创建一个Session，注意到这里的构造又用到了io_context
auto new_session = std::make_shared<Session>(_ioc, this);
//这里_acceptor执行异步的accept函数
//这里做的事情不是accept，而是往io_context中注册一个accept事件，以及这个事件的处理函数
_acceptor.async_accept(
new_session->Socket(), //传入Session的socket，事件发生时，_acceptor会初始化成连接进来的socket
std::bind( //修饰处理收到连接的方法
&Server::handle_accept,
this,
new_session,
std::placeholders::_1
)
);
}
void Server::handle_accept(std::shared_ptr<Session> new_session, const boost::system::error_code& ec) {
if (ec) {
std::cout << "Error occured in accepting. Errmessage: " << ec.message() << std::endl;
//delete new_session;
}
else { //没有错误
//启动Session
new_session->Start();
_sessions.insert(std::make_pair(new_session->GetUuid(), new_session));
}
//这里无论成功还是失败，都要重新注册一个监听事件给io_context
//以便能够继续监听连接
start_accept();
}

复制代码

#pragma once
#include <iostream>
#include <boost/asio.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>
#include "Server.h"
#include <mutex>
#include <queue>
#include "MsgNode.h"
#include <iomanip>
class Server;
class MsgNode;
//注意到这里继承了std::enable_shared_from_this<Session>
//这样可以使得类内部能够使用shared_from_this()统一获取this指针的shared_ptr,shared_ptr的唯一性，防止不同的shared_ptr托管同一个this指针的资源
class Session: public std::enable_shared_from_this<Session>
{
public:
//构造函数
Session(boost::asio::io_context& ioc, Server* server);
boost::asio::ip::tcp::socket& Socket();
//启动读写
void Start();
//发送数据
void Send(char* buf, std::size_t len);
//返回uuid
const std::string& GetUuid() {
return _uuid;
}
~Session() {
std::cout << "delete session: uuid=" << _uuid << std::endl;
}
private:
//处理读事件
void HandleRead(const boost::system::error_code& ec, std::size_t bytes_transferred, std::shared_ptr<Session> _self_ptr);
//处理写事件
void HandleWrite(const boost::system::error_code& ec, std::shared_ptr<Session> _self_ptr);
boost::asio::ip::tcp::socket _socket;
enum {max_length = 1024};
//接受数据的缓冲区
char _data[max_length] = {0};
//发送队列
std::queue<std::shared_ptr<MsgNode>> _send_que;
//队列的锁
std::mutex _send_lock;
//用于标识连接
std::string _uuid;
//引用server，以便能够在读写出错时，清除自身的连接
Server* _server;
//接受消息的节点
std::shared_ptr<MsgNode> _recv_msg_node;
//标识消息头部是否解析完成
bool _b_head_parse;
//头部消息节点
std::shared_ptr<MsgNode> _recv_head_node;
};

复制代码

#include "Session.h"
Session::Session(boost::asio::io_context& ioc, Server* server) : _socket(ioc),
_server(server), _b_head_parse(false) {
//为Session生成uuid
boost::uuids::uuid a_uuid = boost::uuids::random_generator()();
_uuid = boost::uuids::to_string(a_uuid);
//初始化消息头，通常头的长度是一个short类型的长度
_recv_head_node = std::make_shared<MsgNode>(HEAD_LENGTH);
}
//返回使用的socket
boost::asio::ip::tcp::socket& Session::Socket() {
return _socket;
}
void Session::Start() {
//对象创建时已初始化_data
//往io_context注册读事件，区别于另一个async_read函数，这里是能读就返回
_socket.async_read_some(
boost::asio::buffer( //构造一个buffer
_data,
max_length
),
std::bind( //修饰处理
&Session::HandleRead,
this,
std::placeholders::_1,
std::placeholders::_2,
//std::make_shared<Session>(this)导致引用计数不同步
shared_from_this()
)
);
}
void Session::Send(char* buf, std::size_t len) {
bool pending = false;
std::lock_guard<std::mutex> guard(_send_lock);
//如果在为加入消息节点之前，队列不为空，说明这个队列的消息正在发送（写完一个消息后，写事件的处理函数还会继续注册写事件，直到队列为空，所以，如果此时有消息，就说明后续还会继续回调写事件，不再需要在这里注册写事件）
if (_send_que.size() > 0) {
pending = true;
}
//往队列投递消息
_send_que.emplace(std::make_shared<MsgNode>(buf, len));
if (pending) {
return;
}
//注册写事件
boost::asio::async_write(
_socket,
boost::asio::buffer(
buf,
len
),
std::bind(
&Session::HandleWrite,
this,
std::placeholders::_1,
shared_from_this()
)
);
}
//由于这里使用async_read_some，一读到数据就返回，所以处理的过程比较繁琐。还可以可以使用async_read,读取指定字节在返回，其内部实现依然是异步的，和async_read_some一样高效。
void Session::HandleRead(const boost::system::error_code& ec, std::size_t bytes_transferred, std::shared_ptr<Session> _self_ptr) {
if (ec) {
std::cout << "Error occured in reading. ErrMessage: " << ec.message() << std::endl;
_server->ClearSession(_uuid);
}
else {
int copy_len = 0;
while (bytes_transferred > 0) {
if (!_b_head_parse) {
//收到的数据不足头部大小
if (bytes_transferred + _recv_head_node->_cur_len < HEAD_LENGTH) {
memcpy(_recv_head_node->_msg + _recv_head_node->_cur_len, _data + copy_len, bytes_transferred);
_recv_head_node->_cur_len += bytes_transferred;
::memset(_data, 0, max_length);
_socket.async_read_some(boost::asio::buffer(_data, max_length
),
std::bind(&Session::HandleRead, this, std::placeholders::_1, std::placeholders::_2, _self_ptr));
return;
}
//收到的数据比头部多
//头部剩余未复制的长度
int head_remain = HEAD_LENGTH - _recv_head_node->_cur_len;
memcpy(_recv_head_node->_msg + _recv_head_node->_cur_len, _data + copy_len, head_remain);
//更新已处理的data长度和剩余未处理的长度
copy_len += head_remain;
bytes_transferred -= head_remain;
//获取头部数据
int data_len = 0;
memcpy(&data_len, _recv_head_node->_msg, HEAD_LENGTH);
std::cout << "data_len is " << data_len << std::endl;
//头部长度非法
if (data_len > max_length) {
std::cout << "invalid data length is " << data_len << std::endl;
_server->ClearSession(_uuid);
return;
}
_recv_msg_node = std::make_shared<MsgNode>(data_len);
//消息的长度小于头部规定的长度，说明数据未收全，则先将部分消息放到接收节点里
if (bytes_transferred < data_len) {
memcpy(_recv_msg_node->_msg + _recv_msg_node->_cur_len, _data + copy_len, bytes_transferred);
_recv_msg_node->_cur_len += bytes_transferred;
::memset(_data, 0, max_length);
_socket.async_read_some(boost::asio::buffer(_data, max_length),
std::bind(&Session::HandleRead, this, std::placeholders::_1, std::placeholders::_2, _self_ptr));
//头部处理完成
_b_head_parse = true;
return;
}
memcpy(_recv_msg_node->_msg + _recv_msg_node->_cur_len, _data + copy_len, data_len);
_recv_msg_node->_cur_len += data_len;
copy_len += data_len;
bytes_transferred -= data_len;
_recv_msg_node->_msg[_recv_msg_node->_tot_len] = '\0';
std::cout << "receive data is " << _recv_msg_node->_msg << std::endl;
//此处可以调用Send发送测试
Send(_recv_msg_node->_msg, _recv_msg_node->_tot_len);
//继续轮询剩余未处理数据
_b_head_parse = false;
_recv_head_node->Clear();
if (bytes_transferred <= 0) {
::memset(_data, 0, max_length);
_socket.async_read_some(boost::asio::buffer(_data, max_length),
std::bind(&Session::HandleRead, this, std::placeholders::_1, std::placeholders::_2, _self_ptr));
return;
}
continue;
}
//已经处理完头部，处理上次未接受完的消息数据
//接收的数据仍不足剩余未处理的
int remain_msg = _recv_msg_node->_tot_len - _recv_msg_node->_cur_len;
if (bytes_transferred < remain_msg) {
memcpy(_recv_msg_node->_msg + _recv_msg_node->_cur_len, _data + copy_len, bytes_transferred);
_recv_msg_node->_cur_len += bytes_transferred;
::memset(_data, 0, max_length);
_socket.async_read_some(boost::asio::buffer(_data, max_length),
std::bind(&Session::HandleRead, this, std::placeholders::_1, std::placeholders::_2, _self_ptr));
return;
}
memcpy(_recv_msg_node->_msg + _recv_msg_node->_cur_len, _data + copy_len, remain_msg);
_recv_msg_node->_cur_len += remain_msg;
bytes_transferred -= remain_msg;
copy_len += remain_msg;
_recv_msg_node->_msg[_recv_msg_node->_tot_len] = '\0';
std::cout << "receive data is " << _recv_msg_node->_msg << std::endl;
//此处可以调用Send发送测试
Send(_recv_msg_node->_msg, _recv_msg_node->_tot_len);
//继续轮询剩余未处理数据
_b_head_parse = false;
_recv_head_node->Clear();
if (bytes_transferred <= 0) {
::memset(_data, 0, max_length);
_socket.async_read_some(boost::asio::buffer(_data, max_length),
std::bind(&Session::HandleRead, this, std::placeholders::_1, std::placeholders::_2, _self_ptr));
return;
}
continue;
}
}
}
//处理写事件
void Session::HandleWrite(const boost::system::error_code& ec, std::shared_ptr<Session> _self_ptr) {
if (ec) {
std::cout << "Error occured in writting. ErrMessage: " << ec.message() << std::endl;
//delete this;
_server->ClearSession(_uuid);
}
else {
std::lock_guard<std::mutex> guard(_send_lock);
_send_que.pop(); //使用async_write,保证发完指定的字节数
if (!_send_que.empty()) {
auto& msg = _send_que.front();
boost::asio::async_write(
_socket,
boost::asio::buffer(
msg->_msg,
msg->_tot_len
), //使用async_write,可以不做偏移
std::bind(
&Session::HandleWrite,
this,
std::placeholders::_1,
_self_ptr
)
);
}
}
}

复制代码