asio异步读写操作及注意事项

Posted on | In

简介

本文介绍boost asio的异步读写操作及注意事项，为保证知识便于读者吸收，仅介绍api使用的代码片段，下一节再编写完整的客户端和服务器程序。
所以我们定义一个session类，这个session类表示服务器处理客户端连接的管理类

1
2
3
4
5
6
7
class Session {
public:
	Session(std::shared_ptr<asio::ip::tcp::socket> socket);
	void Connect(const asio::ip::tcp::endpoint& ep);
private:
    std::shared_ptr<asio::ip::tcp::socket> _socket;
};

session类定义了一个socket成员变量，负责处理对端的连接读写，封装了Connect函数

1
2
3
void Session::Connect(const asio::ip::tcp::endpoint  &ep) {
	_socket->connect(ep);
}

这里只是简单意思一下，下面核心介绍异步读写api的使用

异步写操作

在写操作前，我们先封装一个Node结构，用来管理要发送和接收的数据，该结构包含数据域首地址，数据的总长度，以及已经处理的长度(已读的长度或者已写的长度)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
//最大报文接收大小
const int RECVSIZE = 1024;
class  MsgNode {
public :
	MsgNode(const char* msg,  int total_len): _total_len(total_len), _cur_len(0){
		_msg = new char[total_len];
		memcpy(_msg, msg, total_len);
	}

	MsgNode(int total_len) :_total_len(total_len), _cur_len(0) {
		_msg = new char[total_len];
	}

	~MsgNode(){
		delete[]_msg;
	}
	//消息首地址
	char* _msg;
	//总长度
	int _total_len;
	//当前长度
	int _cur_len;
};

写了两个构造函数，两个参数的负责构造写节点，一个参数的负责构造读节点。
接下来为Session添加异步写操作和负责发送写数据的节点

1
2
3
4
5
6
7
8
class Session{
public:
	void WriteCallBackErr(const boost::system::error_code & ec, std::size_t bytes_transferred,
    std::shared_ptr<MsgNode>);
	void WriteToSocketErr(const std::string& buf);
private:
    std::shared_ptr<MsgNode> _send_node;
};

WriteToSocketErr函数为我们封装的写操作，WriteCallBackErr为异步写操作回调的函数，为什么会有三个参数呢，
我们可以看一下asio源码

1
2
3
4
5
6
7
8
BOOST_ASIO_COMPLETION_TOKEN_FOR(void (boost::system::error_code,
        std::size_t)) WriteToken
          BOOST_ASIO_DEFAULT_COMPLETION_TOKEN_TYPE(executor_type)>
  BOOST_ASIO_INITFN_AUTO_RESULT_TYPE_PREFIX(WriteToken,
      void (boost::system::error_code, std::size_t))
  async_write_some(const ConstBufferSequence& buffers,
      BOOST_ASIO_MOVE_ARG(WriteToken)token
        BOOST_ASIO_DEFAULT_COMPLETION_TOKEN(executor_type))

async_write_some是异步写的函数，这个异步写函数有两个参数，第一个参数为ConstBufferSequence常引用类型的buffers，
第二个参数为WriteToken类型，而WriteToken在上面定义了，是一个函数对象类型，返回值为void，参数为error_code和size_t，
所以我们为了调用async_write_some函数也要传入一个符合WriteToken定义的函数，就是我们声明的WriteCallBackErr函数，
前两个参数为WriteToken规定的参数，第三个参数为MsgNode的智能指针，这样通过智能指针保证我们发送的Node生命周期延长。
我们看一下WriteToSocketErr函数的具体实现

1
2
3
4
5
6
7
8
void Session::WriteToSocketErr(const std::string& buf) {
	_send_node = make_shared<MsgNode>(buf.c_str(), buf.length());
	//异步发送数据，因为异步所以不会一下发送完
	this->_socket->async_write_some(asio::buffer(_send_node->_msg, 
		_send_node->_total_len),
		std::bind(&Session::WriteCallBackErr,
			this, std::placeholders::_1, std::placeholders::_2, _send_node));
}

因为WriteCallBackErr函数为三个参数且为成员函数，而async_write_some需要的回调函数为两个参数，所以我们通过bind将三个参数转换为两个参数的普通函数。
我们看看回调函数的实现

1
2
3
4
5
6
7
8
9
10
11
void Session::WriteCallBackErr(const boost::system::error_code& ec, 
	std::size_t bytes_transferred, std::shared_ptr<MsgNode> msg_node) {
	if (bytes_transferred + msg_node->_cur_len 
		< msg_node->_total_len) {
		_send_node->_cur_len += bytes_transferred;
		this->_socket->async_write_some(asio::buffer(_send_node->_msg+_send_node->_cur_len,
			_send_node->_total_len-_send_node->_cur_len),
			std::bind(&Session::WriteCallBackErr,
				this, std::placeholders::_1, std::placeholders::_2, _send_node));
	}
}

在WriteCallBackErr函数里判断如果已经发送的字节数没有达到要发送的总字节数，那么久更新节点已经发送的长度，然后计算剩余要发送的长度，如果有数据未发送完，再次调用async_write_some函数异步发送。
但是这个函数并不能投入实际应用，因为async_write_some回调函数返回已发送的字节数可能并不是全部长度。比如TCP发送缓存区总大小为8字节，但是有3字节未发送(上一次未发送完)，这样剩余空间为5字节
https://cdn.llfc.club/1680692232796.jpg
此时我们调用async_write_some发送hello world!实际发送的长度就是为5，也就是只发送了hello，剩余world!通过我们的回调继续发送。
而实际开发的场景用户是不清楚底层tcp的多路复用调用情况的，用户想发送数据的时候就调用WriteToSocketErr,或者循环调用WriteToSocketErr，很可能在一次没发送完数据还未调用回调函数时再次调用WriteToSocketErr，因为boost::asio封装的时epoll和iocp等多路复用模型，当写事件就绪后就发数据，发送的数据按照async_write_some调用的顺序发送，所以回调函数内调用的async_write_some可能并没有被及时调用。
比如我们如下代码

1
2
3
4
//用户发送数据
WriteToSocketErr("Hello World!");
//用户无感知下层调用情况又一次发送了数据
WriteToSocketErr("Hello World!");

那么很可能第一次只发送了Hello，后面的数据没发完，第二次发送了Hello World!之后又发送了World!
所以对端收到的数据很可能是”HelloHello World! World!”
那怎么解决这个问题呢，我们可以通过队列保证应用层的发送顺序。我们在Session中定义一个发送队列，然后重新定义正确的异步发送函数和回调处理

1
2
3
4
5
6
7
8
9
class Session{
public:
    void WriteCallBack(const boost::system::error_code& ec, std::size_t bytes_transferred);
	void WriteToSocket(const std::string &buf);
private:
	std::queue<std::shared_ptr<MsgNode>> _send_queue;
    std::shared_ptr<asio::ip::tcp::socket> _socket;
    bool _send_pending;
};

定义了bool变量_send_pending，该变量为true表示一个节点还未发送完。
_send_queue用来缓存要发送的消息节点，是一个队列。
我们实现异步发送功能

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
39
40
41
42
43
void Session::WriteToSocket(const std::string& buf){
	//插入发送队列
	_send_queue.emplace(new MsgNode(buf.c_str(), buf.length()));
	//pending状态说明上一次有未发送完的数据
	if (_send_pending) {
		return;
	}
	//异步发送数据，因为异步所以不会一下发送完
	this->_socket->async_write_some(asio::buffer(buf), std::bind(&Session::WriteCallBack, this, std::placeholders::_1, std::placeholders::_2));
	_send_pending = true;
}

void Session::WriteCallBack(const boost::system::error_code & ec,  std::size_t bytes_transferred){
	if (ec.value() != 0) {
		std::cout << "Error , code is " << ec.value() << " . Message is " << ec.message();
		return;
	}
	//取出队首元素即当前未发送完数据
	auto & send_data = _send_queue.front();
	send_data->_cur_len += bytes_transferred;
	//数据未发送完， 则继续发送
	if (send_data->_cur_len < send_data->_total_len) {
		this->_socket->async_write_some(asio::buffer(send_data->_msg + send_data->_cur_len, send_data->_total_len-send_data->_cur_len),
			std::bind(&Session::WriteCallBack,
			this, std::placeholders::_1, std::placeholders::_2));
		return;
	}

	//如果发送完，则pop出队首元素
	_send_queue.pop();

	//如果队列为空，则说明所有数据都发送完,将pending设置为false
	if (_send_queue.empty()) {
		_send_pending = false;
	}
	//如果队列不是空，则继续将队首元素发送
	if (!_send_queue.empty()) {
		auto& send_data = _send_queue.front();
		this->_socket->async_write_some(asio::buffer(send_data->_msg + send_data->_cur_len, send_data->_total_len - send_data->_cur_len),
			std::bind(&Session::WriteCallBack,
				this, std::placeholders::_1, std::placeholders::_2));
	}
}

async_write_some函数不能保证每次回调函数触发时发送的长度为要总长度，这样我们每次都要在回调函数判断发送数据是否完成，asio提供了一个更简单的发送函数async_send，这个函数在发送的长度未达到我们要求的长度时就不会触发回调，所以触发回调函数时要么时发送出错了要么是发送完成了,其内部的实现原理就是帮我们不断的调用async_write_some直到完成发送，所以async_send不能和async_write_some混合使用，我们基于async_send封装另外一个发送函数

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
39
//不能与async_write_some混合使用
void Session::WriteAllToSocket(const std::string& buf) {
	//插入发送队列
	_send_queue.emplace(new MsgNode(buf.c_str(), buf.length()));
	//pending状态说明上一次有未发送完的数据
	if (_send_pending) {
		return;
	}
	//异步发送数据，因为异步所以不会一下发送完
	this->_socket->async_send(asio::buffer(buf), 
		std::bind(&Session::WriteAllCallBack, this,
			std::placeholders::_1, std::placeholders::_2));
	_send_pending = true;
}

void Session::WriteAllCallBack(const boost::system::error_code& ec, std::size_t bytes_transferred){
	if (ec.value() != 0) {
		std::cout << "Error occured! Error code = "
			<< ec.value()
			<< ". Message: " << ec.message();

		return;
	}
	
	//如果发送完，则pop出队首元素
	_send_queue.pop();

	//如果队列为空，则说明所有数据都发送完,将pending设置为false
	if (_send_queue.empty()) {
		_send_pending = false;
	}
	//如果队列不是空，则继续将队首元素发送
	if (!_send_queue.empty()) {
		auto& send_data = _send_queue.front();
		this->_socket->async_send(asio::buffer(send_data->_msg + send_data->_cur_len, send_data->_total_len - send_data->_cur_len),
			std::bind(&Session::WriteAllCallBack,
				this, std::placeholders::_1, std::placeholders::_2));
	}
}

异步读操作

接下来介绍异步读操作，异步读操作和异步的写操作类似同样又async_read_some和async_receive函数，前者触发的回调函数获取的读数据的长度可能会小于要求读取的总长度，后者触发的回调函数读取的数据长度等于读取的总长度。
先基于async_read_some封装一个读取的函数ReadFromSocket，同样在Session类的声明中添加一些变量

1
2
3
4
5
6
7
8
9
class Session {
public:
	void ReadFromSocket();
	void ReadCallBack(const boost::system::error_code& ec, std::size_t bytes_transferred);
private:
	std::shared_ptr<asio::ip::tcp::socket> _socket;
	std::shared_ptr<MsgNode> _recv_node;
	bool _recv_pending;
};

_recv_node用来缓存接收的数据，_recv_pending为true表示节点正在接收数据，还未接受完。

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
//不考虑粘包情况， 先用固定的字节接收
void Session::ReadFromSocket() {
	if (_recv_pending) {
		return;
	}
	//可以调用构造函数直接构造，但不可用已经构造好的智能指针赋值
	/*auto _recv_nodez = std::make_unique<MsgNode>(RECVSIZE);
	_recv_node = _recv_nodez;*/
	_recv_node = std::make_shared<MsgNode>(RECVSIZE);
	_socket->async_read_some(asio::buffer(_recv_node->_msg, _recv_node->_total_len), std::bind(&Session::ReadCallBack, this,
		std::placeholders::_1, std::placeholders::_2));
	_recv_pending = true;
}

void Session::ReadCallBack(const boost::system::error_code& ec, std::size_t bytes_transferred){
	_recv_node->_cur_len += bytes_transferred;
	//没读完继续读
	if (_recv_node->_cur_len < _recv_node->_total_len) {
		_socket->async_read_some(asio::buffer(_recv_node->_msg+_recv_node->_cur_len,
			_recv_node->_total_len - _recv_node->_cur_len), std::bind(&Session::ReadCallBack, this,
			std::placeholders::_1, std::placeholders::_2));
		return;
	}
	//将数据投递到队列里交给逻辑线程处理，此处略去
	//如果读完了则将标记置为false
	_recv_pending = false;
	//指针置空
	_recv_node = nullptr;	
}

我们基于async_receive再封装一个接收数据的函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
void Session::ReadAllFromSocket(const std::string& buf) {
	if (_recv_pending) {
		return;
	}
	//可以调用构造函数直接构造，但不可用已经构造好的智能指针赋值
	/*auto _recv_nodez = std::make_unique<MsgNode>(RECVSIZE);
	_recv_node = _recv_nodez;*/
	_recv_node = std::make_shared<MsgNode>(RECVSIZE);
	_socket->async_receive(asio::buffer(_recv_node->_msg, _recv_node->_total_len), std::bind(&Session::ReadAllCallBack, this,
		std::placeholders::_1, std::placeholders::_2));
	_recv_pending = true;
}

void Session::ReadAllCallBack(const boost::system::error_code& ec, std::size_t bytes_transferred) {
	_recv_node->_cur_len += bytes_transferred;
	//将数据投递到队列里交给逻辑线程处理，此处略去
	//如果读完了则将标记置为false
	_recv_pending = false;
	//指针置空
	_recv_node = nullptr;
}

同样async_read_some和async_receive不能混合使用，否则会出现逻辑问题。

源码连接

本文介绍了boost asio异步读写的操作，仅仅是代码片段和api的封装便于大家理解，下一篇利用这些异步api写一个异步的服务器展示收发效果。
源码连接
https://gitee.com/secondtonone1/boostasio-learn