devbox
你好赵伟
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

十、消息头完善和使用json序列化

作者：你好赵伟 | 2024-08-15 09:24:15

赞

踩

十、消息头完善和使用json序列化

        系列文章目录：C++ asio网络编程-CSDN博客 

        本教程使用的是jsoncpp，安装方式网上有很多，可以自己下载源码编译，也可以使用vcpkg一键安装，我比较推荐使用vcpkg，感觉这个就像 visual studio 的 maven，使用这种方式安装后就不需要自己配置项目的包含目录和库目录了，直接使用。

1、消息头完善

        前面我们的消息头其实是一种简化的方式去构造的，从本小节开始将使用完整的 tlv 格式

2、json数据格式

        json中将id和data分开存放：

  1. {
  2.         "data" : "hello",
  3.         "id" : 1001
  4. }

3、MsgNode修改       

        现在，我们修改一下MsgNode

  1. #pragma once
  2. #include <string>
  3. #include <iostream>
  4. #include <boost/asio.hpp>
  5. #include "const.h"
  6.  
  7. class MsgNode
  8. {
  9. public:
  10. 	MsgNode(short max_len);
  11. 	~MsgNode();
  12.  
  13. 	void clearData();  // 清空数据
  14.  
  15. public:
  16. 	short _cur_len;  // 当前收发数据的长度
  17. 	short _total_len; // 数据总长度
  18. 	char* _data;  // 数据起始地址
  19. };
  20.  
  21.  
  22. class RecvNode : public MsgNode
  23. {
  24. public:
  25. 	RecvNode(short max_len, short msg_id);
  26.  
  27. private:
  28. 	short _msg_id;
  29. };
  30.  
  31. class SendNode : public MsgNode
  32. {
  33. public:
  34. 	SendNode(const char* msg, short max_len, short msg_id);
  35.  
  36. private:
  37. 	short _msg_id;
  38. };
  39.  
  1. #include "MsgNode.h"
  2.  
  3. MsgNode::MsgNode(short max_len) : _total_len(max_len), _cur_len(0)
  4. {
  5. 	_data = new char[_total_len + 1]();
  6. 	_data[_total_len] = '\0';
  7. }
  8.  
  9. MsgNode::~MsgNode() {
  10. 	std::cout << "destruct MsgNode" << std::endl;
  11. 	delete[] _data;
  12. }
  13.  
  14. void MsgNode::clearData()
  15. {
  16. 	memset(_data, 0, _total_len);
  17. 	_cur_len = 0;
  18. }
  19.  
  20. RecvNode::RecvNode(short max_len, short msg_id)
  21. 	: MsgNode(max_len), _msg_id(msg_id)
  22. {
  23. }
  24.  
  25. SendNode::SendNode(const char* msg, short max_len, short msg_id)
  26. 	: MsgNode(max_len + HEAD_TOTAL_LEN), _msg_id(msg_id)
  27. {
  28. 	// 先拼接id
  29. 	short msg_id_host = boost::asio::detail::socket_ops::host_to_network_short(msg_id);
  30. 	memcpy(_data, &msg_id_host, HEAD_ID_LEN);
  31. 	// 再拼接长度
  32. 	short msg_len_host = boost::asio::detail::socket_ops::host_to_network_short(max_len);
  33. 	memcpy(_data + HEAD_ID_LEN, &msg_len_host, HEAD_DATA_LEN);
  34. 	// 最后拼接消息
  35. 	memcpy(_data + HEAD_TOTAL_LEN, msg, max_len);
  36. }

        const.h的内容如下

  1. #pragma once
  2.  
  3. #define MAX_LENGTH  1024*2
  4. #define MAX_SENDQUE 1000
  5. #define MAX_RECVQUE 10000
  6. #define HEAD_TOTAL_LEN 4
  7. #define HEAD_ID_LEN 2
  8. #define HEAD_DATA_LEN 2

4、Session类修改      

  1. #pragma once
  2. #include <iostream>
  3. #include <boost/asio.hpp>
  4. #include <map>
  5. #include "Server.h"
  6. #include <queue>
  7. #include <mutex>
  8. #include "MsgNode.h"
  9.  
  10. class Server;
  11.  
  12. class Session : public std::enable_shared_from_this<Session>
  13. {
  14. public:
  15. 	Session(boost::asio::io_context& ioc, Server* server);
  16. 	~Session();
  17.  
  18. 	void start();
  19. 	void close();
  20.  
  21. 	void send(char* msg, int max_length, short msg_id);
  22. 	void send(std::string msg, short msg_id);
  23.  
  24. 	boost::asio::ip::tcp::socket& getSocket();
  25. 	std::string& getUuid();
  26. 	std::shared_ptr<Session> getSelfShared();
  27.  
  28. private:
  29. 	void handle_read(const boost::system::error_code& ec,
  30. 		std::size_t bytes_transferred,
  31. 		std::shared_ptr<Session> self_share);
  32. 	void handle_write(const boost::system::error_code& ec, 
  33. 		std::shared_ptr<Session> self_share);
  34.  
  35. 	boost::asio::ip::tcp::socket _socket;
  36. 	std::string _uuid;
  37. 	Server* _server;
  38. 	bool _b_close;
  39.  
  40. 	char _data[MAX_LENGTH];
  41.  
  42. 	// 收到的消息头
  43. 	std::shared_ptr<MsgNode> _recv_head_node;
  44. 	// 消息头是否处理完成
  45. 	bool _b_head_parse;
  46. 	// 收到的消息
  47. 	std::shared_ptr<RecvNode> _recv_msg_node;
  48.  
  49. 	// 发送的消息
  50. 	std::queue<std::shared_ptr<SendNode>> _send_que;
  51. 	std::mutex _send_lock;
  52. };
  53.  
  54.  
  55.  
  56.  
  1. #include "Session.h"
  2. #include <boost/uuid/uuid_generators.hpp>
  3. #include <boost/uuid/uuid_io.hpp>
  4. #include <json/json.h>
  5. #include <json/value.h>
  6. #include <json/reader.h>
  7.  
  8. Session::Session(boost::asio::io_context& ioc, Server* server)
  9. 	: _socket(ioc), _server(server), _b_close(false), _b_head_parse(false)
  10. {
  11. 	// 生成唯一id，可以了解一下雪花算法
  12. 	// 这里直接使用boost自带的方法
  13. 	boost::uuids::uuid a_uuid = boost::uuids::random_generator()();
  14. 	_uuid = boost::uuids::to_string(a_uuid);
  15.  
  16. 	_recv_head_node = std::make_shared<MsgNode>(HEAD_TOTAL_LEN);
  17.  
  18. }
  19.  
  20. Session::~Session()
  21. {
  22. 	std::cout << "~Session destruct" << std::endl;
  23. }
  24.  
  25. boost::asio::ip::tcp::socket& Session::getSocket()
  26. {
  27. 	return _socket;
  28. }
  29.  
  30. void Session::start()
  31. {
  32. 	memset(_data, 0, MAX_LENGTH);
  33. 	_socket.async_read_some(boost::asio::buffer(_data, MAX_LENGTH),
  34. 		std::bind(&Session::handle_read, this, 
  35. 			std::placeholders::_1, std::placeholders::_2,
  36. 			getSelfShared()));
  37. }
  38.  
  39. void Session::send(char* msg, int max_length, short msg_id)
  40. {
  41. 	std::lock_guard<std::mutex> lock(_send_lock);
  42.  
  43. 	if (_send_que.size() > MAX_SENDQUE) {
  44. 		std::cout << "session: " << _uuid << " send que fulled, size is " << MAX_SENDQUE << std::endl;
  45. 		return;
  46. 	}
  47.  
  48. 	_send_que.push(std::make_shared<SendNode>(msg, max_length, msg_id));
  49. 	if (_send_que.size() > 1) {  // 因为push了一次，判断条件为大于1
  50. 		return;
  51. 	}
  52.  
  53. 	auto& msgnode = _send_que.front();
  54. 	boost::asio::async_write(_socket, boost::asio::buffer(msgnode->_data, msgnode->_total_len),
  55. 		std::bind(&Session::handle_write, this, std::placeholders::_1, getSelfShared()));
  56. }
  57.  
  58. void Session::send(std::string msg, short msg_id)
  59. {
  60. 	std::lock_guard<std::mutex> lock(_send_lock);
  61.  
  62. 	if (_send_que.size() > MAX_SENDQUE) {
  63. 		std::cout << "session: " << _uuid << " send que fulled, size is " << MAX_SENDQUE << std::endl;
  64. 		return;
  65. 	}
  66.  
  67. 	_send_que.push(std::make_shared<SendNode>(msg.c_str(), msg.length(), msg_id));
  68. 	if (_send_que.size() > 1) {  // 因为push了一次，判断条件为大于1
  69. 		return;
  70. 	}
  71.  
  72. 	auto& msgnode = _send_que.front();
  73. 	boost::asio::async_write(_socket, boost::asio::buffer(msgnode->_data, msgnode->_total_len),
  74. 		std::bind(&Session::handle_write, this, std::placeholders::_1, getSelfShared()));
  75. }
  76.  
  77. void Session::close()
  78. {
  79. 	_socket.close();
  80. 	_b_close = true;
  81. }
  82.  
  83. std::string& Session::getUuid()
  84. {
  85. 	return _uuid;
  86. }
  87.  
  88. std::shared_ptr<Session> Session::getSelfShared()
  89. {
  90. 	return shared_from_this();
  91. }
  92.  
  93. void Session::handle_read(const boost::system::error_code& ec, 
  94. 	std::size_t bytes_transferred, std::shared_ptr<Session> self_share)
  95. {
  96. 	try {
  97. 		if (!ec) {
  98. 			int copy_len = 0;  // 已经处理的字符数
  99. 			while (bytes_transferred > 0) {
  100. 				// 头部未处理完成
  101. 				if (!_b_head_parse) {
  102. 					// 接收到的数据长度比头部长度小
  103. 					if (bytes_transferred + _recv_head_node->_cur_len < HEAD_TOTAL_LEN) {
  104. 						// 将接收到的存起来
  105. 						memcpy(_recv_head_node->_data + _recv_head_node->_cur_len, _data + copy_len, bytes_transferred);
  106. 						_recv_head_node->_cur_len += bytes_transferred;
  107. 						// 继续接收数据
  108. 						memset(_data, 0, MAX_LENGTH);
  109. 						_socket.async_read_some(boost::asio::buffer(_data, MAX_LENGTH),
  110. 							std::bind(&Session::handle_read, this, std::placeholders::_1, std::placeholders::_2, self_share));
  111. 						return;
  112. 					}
  113. 					// 接收到的长度大于等于头部长度
  114. 					// 只存头部剩余部分
  115. 					int head_remain = HEAD_TOTAL_LEN - _recv_head_node->_cur_len;
  116. 					memcpy(_recv_head_node->_data + _recv_head_node->_cur_len, _data + copy_len, head_remain);
  117. 					// 更新已处理的data长度和接收到但未处理的长度
  118. 					copy_len += head_remain;
  119. 					bytes_transferred -= head_remain;
  120. 					// 获取头部 msg_id 数据
  121. 					short msg_id = 0;
  122. 					memcpy(&msg_id, _recv_head_node->_data, HEAD_ID_LEN);
  123. 					// 网络字节序转本地字节序
  124. 					msg_id = boost::asio::detail::socket_ops::network_to_host_short(msg_id);
  125. 					std::cout << "msg id: " << msg_id << std::endl;
  126. 					// id 校验
  127. 					if (msg_id > MAX_LENGTH) {
  128. 						std::cout << "invalid msd_id: " << msg_id << std::endl;
  129. 						_server->clearSession(_uuid);
  130. 						return;
  131. 					}
  132. 					// 获取数据长度
  133. 					short msg_len = 0;
  134. 					memcpy(&msg_len, _recv_head_node->_data + HEAD_ID_LEN, HEAD_DATA_LEN);
  135. 					// 网络字节序转化为本地字节序
  136. 					msg_len = boost::asio::detail::socket_ops::network_to_host_short(msg_len);
  137. 					std::cout << "msg len: " << msg_len << std::endl;
  138. 					// 头部长度非法
  139. 					if (msg_len > MAX_LENGTH) {
  140. 						std::cout << "invalid data length" << std::endl;
  141. 						_server->clearSession(_uuid);
  142. 						return;
  143. 					}
  144. 					// 头部处理完成
  145. 					_b_head_parse = true;
  146. 					// 头部处理完成后，还有剩余接收到的数据，就是正式数据
  147. 					// 开始处理数据
  148. 					_recv_msg_node = std::make_shared<RecvNode>(msg_len, msg_id);
  149. 					// 有数据但不完整
  150. 					if (bytes_transferred < msg_len) {
  151. 						// 先存下已接收到的
  152. 						memcpy(_recv_msg_node->_data + _recv_msg_node->_cur_len, _data + copy_len, bytes_transferred);
  153. 						_recv_msg_node->_cur_len += bytes_transferred;
  154. 						// 继续接收剩下的数据
  155. 						memcpy(_data, 0, MAX_LENGTH);
  156. 						_socket.async_read_some(boost::asio::buffer(_data, MAX_LENGTH),
  157. 							std::bind(&Session::handle_read, this, std::placeholders::_1, std::placeholders::_2, self_share));
  158. 						return;
  159. 					}
  160. 					// 数据齐全
  161. 					memcpy(_recv_msg_node->_data + _recv_msg_node->_cur_len, _data + copy_len, msg_len);
  162. 					_recv_msg_node->_cur_len += msg_len;
  163. 					copy_len += msg_len;
  164. 					bytes_transferred -= msg_len;
  165. 					_recv_msg_node->_data[_recv_msg_node->_total_len] = '\0';
  166. 					std::cout << "receive data: " << _recv_msg_node->_data << std::endl;
  167.  
  168. 					// 发过来的是json，解析数据测试是否正确
  169. 					Json::Reader reader;
  170. 					Json::Value root;
  171. 					reader.parse(std::string(_recv_msg_node->_data, _recv_msg_node->_total_len), root);
  172. 					std::cout << "receive msg id is " << root["id"].asInt() << " msg data is "
  173. 						<< root["data"].asString() << std::endl;
  174. 					// 发送测试
  175. 					root["data"] = "server has received msg, msg data is " + root["data"].asString();
  176. 					std::string return_str = root.toStyledString();  // 序列化后再发送
  177. 					send(return_str, root["id"].asInt());
  178.  
  179. 					// 继续轮询剩余未处理的数据
  180. 					_b_head_parse = false;
  181. 					_recv_head_node->clearData();
  182. 					if (bytes_transferred <= 0) {
  183. 						memset(_data, 0, MAX_LENGTH);
  184. 						_socket.async_read_some(boost::asio::buffer(_data, MAX_LENGTH),
  185. 							std::bind(&Session::handle_read, this, std::placeholders::_1, std::placeholders::_2, self_share));
  186. 						return;
  187. 					}
  188. 					continue;
  189. 				}
  190.  
  191. 				// 已经处理完头部，继续处理未处理完的数据
  192. 				int remain_msg = _recv_msg_node->_total_len - _recv_msg_node->_cur_len;
  193. 				// 接收的数据仍不完整
  194. 				if (bytes_transferred < remain_msg) {
  195. 					memcpy(_recv_msg_node->_data + _recv_msg_node->_cur_len, _data + copy_len, bytes_transferred);
  196. 					_recv_msg_node->_cur_len += bytes_transferred;
  197. 					// 继续接收数据
  198. 					memset(_data, 0, MAX_LENGTH);
  199. 					_socket.async_read_some(boost::asio::buffer(_data, MAX_LENGTH),
  200. 						std::bind(&Session::handle_read, this, std::placeholders::_1, std::placeholders::_2, self_share));
  201. 					return;
  202. 				}
  203. 				// 接受的数据完整
  204. 				memcpy(_recv_msg_node->_data + _recv_msg_node->_cur_len, _data + copy_len, remain_msg);
  205. 				_recv_msg_node->_cur_len += remain_msg;
  206. 				bytes_transferred -= remain_msg;
  207. 				copy_len += remain_msg;
  208. 				_recv_msg_node->_data[_recv_msg_node->_total_len] = '\0';
  209. 				//std::cout << "receive data: " << _recv_msg_node->_data << std::endl;
  210.  
  211. 				// 发过来的是json，解析数据测试是否正确
  212. 				Json::Reader reader;
  213. 				Json::Value root;
  214. 				reader.parse(std::string(_recv_msg_node->_data, _recv_msg_node->_total_len), root);
  215. 				std::cout << "receive msg id is " << root["id"].asInt() << " msg data is "
  216. 					<< root["data"].asString() << std::endl;
  217. 				// 发送测试
  218. 				root["data"] = "server has received msg, msg data is " + root["data"].asString();
  219. 				std::string return_str = root.toStyledString();  // 序列化后再发送
  220. 				send(return_str, root["id"].asInt());
  221.  
  222. 				// 继续轮询剩余未处理的数据
  223. 				_b_head_parse = false;
  224. 				_recv_head_node->clearData();
  225. 				if (bytes_transferred <= 0) {
  226. 					memset(_data, 0, MAX_LENGTH);
  227. 					_socket.async_read_some(boost::asio::buffer(_data, MAX_LENGTH),
  228. 						std::bind(&Session::handle_read, this, std::placeholders::_1, std::placeholders::_2, self_share));
  229. 					return;
  230. 				}
  231. 				continue;
  232. 			}
  233. 		}
  234. 		else {
  235. 			std::cout << "read error" << std::endl;
  236. 			close();
  237. 			_server->clearSession(_uuid);
  238. 		}
  239. 	}
  240. 	catch (std::exception& e) {
  241. 		std::cout << "Exception: " << e.what() << std::endl;
  242. 	}
  243. }
  244.  
  245. void Session::handle_write(const boost::system::error_code& ec, 
  246. 	std::shared_ptr<Session> self_share)
  247. {
  248. 	try {
  249. 		if (!ec) {
  250. 			std::lock_guard<std::mutex> lock(_send_lock);
  251. 			_send_que.pop();
  252. 			if (!_send_que.empty()) {
  253. 				auto& msgNode = _send_que.front();
  254. 				boost::asio::async_write(_socket, boost::asio::buffer(msgNode->_data, msgNode->_total_len),
  255. 					std::bind(&Session::handle_write, this, std::placeholders::_1, self_share));
  256. 			}
  257. 		}
  258. 		else {
  259. 			std::cout << "write error: " << ec.what() << std::endl;
  260. 			close();
  261. 			_server->clearSession(_uuid);
  262. 		}
  263. 	}
  264. 	catch (std::exception& e) {
  265. 		std::cerr << "Exception: " << e.what() << std::endl;
  266. 	}
  267. }

5、客户端修改

  1. #include <boost/asio.hpp>
  2. #include <iostream>
  3. #include <thread>
  4. #include <json/json.h>
  5. #include <json/value.h>
  6. #include <json/reader.h>
  7.  
  8. #define MAX_LENGTH  1024*2
  9. #define HEAD_LENGTH 2
  10. #define HEAD_TOTAL 4
  11.  
  12. int main() {
  13.  
  14. 	try {
  15. 		// 创建上下文服务
  16. 		boost::asio::io_context ioc;
  17. 		// 构造endpoint
  18. 		boost::asio::ip::tcp::endpoint remote_ep(boost::asio::ip::address::from_string("127.0.0.1"), 10086);
  19. 		boost::asio::ip::tcp::socket sock(ioc);
  20. 		boost::system::error_code error = boost::asio::error::host_not_found;
  21. 		sock.connect(remote_ep, error);
  22. 		if (error) {
  23. 			std::cout << "连接失败，错误码：" << error.value()
  24. 				<< "错误信息：" << error.message() << std::endl;
  25. 			return 0;
  26. 		}
  27.  
  28. 		Json::Value root;
  29. 		root["id"] = 1001;
  30. 		root["data"] = "hello";
  31. 		std::string request = root.toStyledString();
  32. 		std::size_t request_length = request.length();
  33. 		char send_data[MAX_LENGTH] = { 0 };
  34. 		int msgid = 1001;
  35. 		int msgid_host = boost::asio::detail::socket_ops::host_to_network_short(msgid);
  36. 		memcpy(send_data, &msgid_host, 2);
  37. 		int request_host_length = boost::asio::detail::socket_ops::host_to_network_short(request_length);
  38. 		memcpy(send_data + 2, &request_host_length, 2);
  39. 		memcpy(send_data + 4, request.c_str(), request_length);
  40. 		boost::asio::write(sock, boost::asio::buffer(send_data, request_length + 4));
  41. 		std::cout << "begin to receive ..." << std::endl;
  42.  
  43. 		char reply_head[HEAD_TOTAL];
  44. 		std::size_t reply_length = boost::asio::read(sock, boost::asio::buffer(reply_head, HEAD_TOTAL));
  45. 		msgid = 0;
  46. 		memcpy(&msgid, reply_head, 2);
  47. 		short msglen = 0;
  48. 		memcpy(&msglen, reply_head + 2, 2);
  49.  
  50. 		// 转换为本地字节序
  51. 		msgid = boost::asio::detail::socket_ops::network_to_host_short(msgid);
  52. 		msglen = boost::asio::detail::socket_ops::network_to_host_short(msglen);
  53.  
  54. 		char msg[MAX_LENGTH] = { 0 };
  55. 		std::size_t msg_length = boost::asio::read(sock, boost::asio::buffer(msg, msglen));
  56. 		Json::Reader reader;
  57. 		reader.parse(std::string(msg, msg_length), root);
  58. 		std::cout << "msg id: " << root["id"] << ", msg: " << root["data"] << std::endl;
  59.  
  60. 		getchar();
  61. 	}
  62. 	catch (std::exception& e) {
  63. 		std::cout << "异常：" << e.what() << std::endl;
  64. 	}
  65.  
  66. 	return 0;
  67. }

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号