CSession#

在ChatRoom中，CSession类是一个重要的组件，负责处理客户端连接和消息交互。它通常包含以下功能：

消息处理：接收和发送消息。
连接管理：处理客户端的连接和断开连接事件。
状态管理：维护会话的状态信息，例如用户ID、会话ID等。

CSession会在服务器端创建一个会话对象(io_context, socket)来代表每个连接的客户端。存储在服务器CServer类中，通常会有一个会话列表std::map<std::string _uuid, std::shared_ptr<CSession>>来管理所有活跃的会话。

CSession类定义#

1
#pragma once
2
#include <boost/asio.hpp>
3
#include <boost/uuid/uuid.hpp>
4
#include <boost/uuid/uuid_generators.hpp>
5
#include <boost/uuid/uuid_io.hpp>
6
#include <queue>
7
#include "Const.h"
8
#include "MsgNode.h"
9
#include "CServer.h"
10

11
/*
12
CSession 类用于表示一个网络会话，封装了与客户端的连接、数据接收和发送等功能。
13

14
功能解析：
15
1. 构造函数和析构函数：
16
   - CSession()：初始化会话，设置默认值。
17
   - ~CSession()：清理资源，关闭连接。
18
2. 获取套接字和UUID：
19
   - getSocket()：返回与客户端的套接字引用。
20
   - getUuid()：返回会话的唯一标识符。
21
3. 会话管理：
22
   - start()：开始会话，通常用于启动异步操作。
23
   - close()：关闭会话，释放资源。
24
4. 异步操作：
25
   - asyncReadHead()：异步读取数据包头部，指定总长度。
26
   - asyncReadBody()：异步读取数据包体，指定总长度。
27
   - send()：发送数据到客户端。
28
5. 数据处理：
29
   - asyncReadFull()：异步读取完整数据包。
30
   - asyncReadLen()：异步读取指定长度的数据。
31

32
*/
33

34
class CSession : public std::enable_shared_from_this<CSession>{
35
public:
36
  CSession(boost::asio::io_context& io_context, CServer* server);
37
  ~CSession();
38
  boost::asio::ip::tcp::socket& getSocket();
39
  std::string& getUuid();
40
  void start();
41
  void close();
42
  // 异步解析头部
43
  void asyncReadHead(int total_length);
44
  // 解析并处理数据包体
45
  void asyncReadBody(int total_length);
46
  // send data
47
  void send(const char* msg, short max_length, short msg_id);
48
  void send(const std::string& msg, short msg_id);
49
  // 支持右值引用，临时字符串对象传递。
50
  void send(std::string&& msg, short msg_id);
51

52
private:
53
  // 解析完整的数据包(head + body)
54
  // max_length表示头部长度
55
  // handler为回调函数
56
  void asyncReadFull(std::size_t max_length, std::function<void(const boost::system::error_code& error, std::size_t)> handler);
57
  // 封装async_read_some异步读取函数
58
  // 读取指定长度的数据，read_length为已处理的数据，total_length为包的总长度。
59
  void asyncReadLen(std::size_t read_length, std::size_t total_length, std::function<void(const boost::system::error_code& error, std::size_t)> handler);
60

61
  void handleWrite(const boost::system::error_code& error, std::shared_ptr<CSession> self_shared);
62

63
  // server
64
  boost::asio::ip::tcp::socket _socket;
65
  CServer* _server;
66
  bool _b_close;
67

68
  // session uid
69
  std::string _uuid;
70

71
  // data, 单个消息的缓冲区
72
  char _data[MAX_LENGTH];
73

74
  // send data
75
  std::queue<std::shared_ptr<SendNode>> _send_queue;
76
  std::mutex _send_lock;
77

78

79
  bool _b_head_parse;
80
  // head data node
81
  std::shared_ptr<MsgNode> _recv_head_node;
82
  // msg data node
83
  std::shared_ptr<RecvNode> _recv_msg_node;
84

85
};
86

87
class LogicNode {
88
  friend class LogicSystem;
89
public:
90
  LogicNode(std::shared_ptr<CSession>, std::shared_ptr<RecvNode>);
91
private:
92
  std::shared_ptr<CSession> _session;
93
  std::shared_ptr<RecvNode> _recv_node;
94
};

异步消息处理#

消息接收#

CSession类使用Boost.Asio库来处理异步IO操作。它通过asyncReadHead()和asyncReadBody()方法来异步读取数据包的头部和体。

asyncReadHead()和asyncReadBody()方法会调用asyncReadFull()来读取指定长度的数据。这样可以确保在处理大数据包时不会阻塞IO线程。

asyncReadFull()方法会进一步调用asyncReadLen()来异步读取数据，asyncReadLen()会调用asio中的async_read_some()函数，异步的读取指定长度的数据到缓冲区_data，并在读取完成后调用指定的回调函数。

在回调函数中，会解析缓冲区_data中的数据包头部和体的数据，并将其存储在_recv_head_node和_recv_msg_node中。

随后_recv_msg_node会被传递给LogicSystem，由逻辑系统处理数据。

代码实现#

1
#include "CSession.h"
2
#include <iostream>
3
#include "LogicSystem.h"
4

5
void CSession::asyncReadFull(std::size_t max_length, std::function<void(const boost::system::error_code& error, std::size_t)> handler)
6
{
7
  ::memset(_data, 0, max_length);
8
  asyncReadLen(0, max_length, handler);
9
}
10

11
void CSession::asyncReadLen(std::size_t read_length, std::size_t total_length, std::function<void(const boost::system::error_code& error, std::size_t)> handler) {
12
  auto self = shared_from_this();
13
  _socket.async_read_some(boost::asio::buffer(_data + read_length, total_length - read_length),
14
    [self, read_length, total_length, handler](const boost::system::error_code& error, std::size_t bytes_transfered) {
15
      // 出错，错误交给回调函数处理。
16
      if (error) {
17
        handler(error, read_length + bytes_transfered);
18
        return;
19
      }
20
      // 读取到的长度足够了，回调函数
21
      if (read_length + bytes_transfered >= total_length) {
22
        handler(error, read_length + bytes_transfered);
23
        return;
24
      }
25

26
      self->asyncReadLen(read_length + bytes_transfered, total_length, handler);
27

28

29
    });
30
}
31

32
void CSession::asyncReadHead(int total_len)
33
{
34
  auto self = shared_from_this();
35
  asyncReadFull(HEAD_TOTAL_LENGTH, [self, this](const boost::system::error_code& error, std::size_t bytes_transfered) {
36
    try {
37
      if (error) {
38
        std::cout << "handle read failed, error is: " << error.what() << std::endl;
39
        close();
40
        _server->clearSession(_uuid);
41
        return;
42
      }
43

44
      if (bytes_transfered < HEAD_TOTAL_LENGTH) {
45
        std::cout << "read head length not match, read [" << bytes_transfered << "] , total[" << HEAD_TOTAL_LENGTH << "] " << std::endl;
46
        close();
47
        _server->clearSession(_uuid);
48
        return;
49
      }
50

51
      _recv_head_node->clear();
52
      memcpy(_recv_head_node->_data, _data, bytes_transfered);
53

54
      // get MSGID
55
      short msg_id = 0;
56
      memcpy(&msg_id, _recv_head_node->_data, HEAD_ID_LENGTH);
57
      msg_id = boost::asio::detail::socket_ops::network_to_host_short(msg_id);
58
      std::cout << "msg id is " << msg_id;
59

60
      // msg_id 非法
61
      if (msg_id > MAX_LENGTH) {
62
        std::cout << "msg_id too long" << std::endl;
63
        _server->clearSession(_uuid);
64
        return;
65
      }
66

67
      short msg_len = 0;
68
      memcpy(&msg_len, _recv_head_node->_data + HEAD_ID_LENGTH, HEAD_DATA_LENGTH);
69
      msg_len = boost::asio::detail::socket_ops::network_to_host_short(msg_len);
70
      std::cout << "msg_len is" << msg_len << std::endl;
71
      if (msg_len > MAX_LENGTH) {
72
        std::cout << "msg_len too long" << std::endl;
73
        _server->clearSession(_uuid);
74
        return;
75
      }
76

77
      // 构造RecvNode，开始读取消息体
78
      _recv_msg_node = std::make_shared<RecvNode>(msg_len, msg_id);
79
      asyncReadBody(msg_len);
80

81
    }
82
    catch (const std::exception& e) {
83
      // Handle exception
84
      std::cerr << "Exception in asyncReadHead: " << e.what() << std::endl;
85
    }
86
    });
87
}
88

89
void CSession::asyncReadBody(int total_length) {
90
  auto self = shared_from_this();
91
  asyncReadFull(total_length, [self, this, total_length](const boost::system::error_code& error, std::size_t bytes_transfered) {
92
    try {
93
      if (error) {
94
        std::cout << "handle read failed, error is " << error.what() << std::endl;
95
        close();
96
        _server->clearSession(_uuid);
97
        return;
98
      }
99
      if (bytes_transfered < total_length) {
100
        std::cout << "read length not match read[" << bytes_transfered << "], total[" << total_length << "]" << std::endl;
101
        close();
102
        _server->clearSession(_uuid);
103
        return;
104
      }
105

106
      _recv_msg_node->clear();
107
      memcpy(_recv_msg_node->_data, _data + HEAD_TOTAL_LENGTH, bytes_transfered);
108
      _recv_msg_node->_cur_length += bytes_transfered;
109
      _recv_msg_node->_data[_recv_msg_node->_total_length] = '\0';
110
      std::cout << "recv_msg_data is" << _recv_msg_node->_data << std::endl;
111

112
      // 构造LogicNode，放入消息队列
113
      LogicSystem::getInstance()->PostMsgToQueue(std::make_shared<LogicNode>(shared_from_this(), _recv_msg_node));
114

115
            // 当前消息处理完成，继续读取下一个消息头
116
      asyncReadHead(HEAD_TOTAL_LENGTH);
117

118
    }
119
    catch (std::exception& e) {
120
      std::cerr << "Exception code is" << e.what() << std::endl;
121
    }
122

123
  });
124
}
125

126
LogicNode::LogicNode(std::shared_ptr<CSession> session, std::shared_ptr<RecvNode> recv_node): _session(session), _recv_node(recv_node)
127
{
128

129
}

消息发送#

当LogicSystem处理完数据后，会调用CSession::send()方法来发送数据。该方法会将待发送的消息放入一个队列中，并使用异步IO操作来发送数据。

这发送队列_send_queue是一个线程安全的队列，使用std::mutex来保护对队列的访问。在CSession和LogicSystem之间构建了一个生产者-消费者模型，CSession负责发送数据，LogicSystem负责产生数据。

每次发送数据时，CSession会检查发送队列是否为空，如果队列只有1个消息，则从队列中取出一个消息并发送。如果大于1个消息，则将所有消息放入队列中，等待发送。这样可以确保始终有且只有一个消息在执行发送操作。

数据发送通过boost::asio::async_write()方法实现。该方法会将数据异步写入到套接字中，并在写入完成后调用指定的回调函数。

在回调函数中重复检查发送队列是否为空，如果不为空，则再次调用boost::asio::async_write()来发送下一个消息。

代码实现#

1
void CSession::handleWrite(const boost::system::error_code& error, std::shared_ptr<CSession> self_shared)
2
{
3
  try {
4
    if(error){
5
      std::cout << "handle write failed, error is " << error.what() << std::endl;
6
      close();
7
      _server->clearSession(_uuid);
8
    }
9
    std::lock_guard<std::mutex> lock(_send_lock);
10
    _send_queue.pop();
11
    if (!_send_queue.empty()) {
12
      auto& send_node = _send_queue.front();
13
      boost::asio::async_write(_socket, boost::asio::buffer(send_node->_data, send_node->_total_length), std::bind(&CSession::handleWrite, this, std::placeholders::_1, self_shared));
14
    }
15
  }
16
  catch (std::exception& e) {
17
    std::cout << "Exception code: " << e.what() << std::endl;
18
  }
19
}
20

21
void CSession::send(const char* msg, short max_length, short msg_id) {
22
  std::lock_guard<std::mutex> lock(_send_lock);
23
  int send_queue_size = _send_queue.size();
24
  if (_send_queue.size() >= MAX_SEND_QUEUE) {
25
    std::cout << "session:" << _uuid << ", send queue is full" << std::endl;
26
    return;
27
  }
28
  _send_queue.push(std::make_shared<SendNode>(msg, max_length, msg_id));
29
  if (_send_queue.size() > 1) {
30
    return;
31
  }
32
  auto& send_node = _send_queue.front();
33
  boost::asio::async_write(_socket, boost::asio::buffer(send_node->_data, send_node->_total_length),
34
    std::bind(&CSession::handleWrite, this, std::placeholders::_1, shared_from_this()));
35

36
}
37

38
void CSession::send(const std::string& msg, short msg_id) {
39
  // 使用 c_str() 转换为 const char*
40
  send(msg.c_str(), static_cast<short>(msg.length()), msg_id);
41
}
42

43
// 支持移动语义的重载（C++11及以上）
44
void CSession::send(std::string&& msg, short msg_id) {
45
  send(msg.c_str(), static_cast<short>(msg.length()), msg_id);
46
}

MessageNode的实现#

1
#pragma once
2
#include "Const.h"
3
class MsgNode{
4
public:
5
  MsgNode(int max_len): _total_length(max_len), _cur_length(0) {
6
    _data = new char[_total_length + 1]();
7
    _data[_total_length] = '\0';
8
  }
9
  void clear();
10
  char* _data;
11
  int _total_length;
12
  int _cur_length;
13
};
14

15
class RecvNode : public MsgNode{
16
public:
17
  RecvNode(int total_length, int msg_id) : MsgNode(total_length), _msg_id(msg_id){
18

19
  }
20
private:
21
  int _msg_id;
22
};
23

24
class SendNode : public MsgNode {
25
public:
26
  SendNode(const char* msg, int max_length, int msg_id):
27
    MsgNode(max_length + HEAD_TOTAL_LENGTH), _msg_id(msg_id){
28
    // 发送前将头部数据转化为大端序。
29
    short host_msg_id = boost::asio::detail::socket_ops::host_to_network_short(_msg_id);
30
    memcpy(_data, &host_msg_id, HEAD_ID_LENGTH);
31
    short host_max_length = boost::asio::detail::socket_ops::host_to_network_short(_total_length);
32
    memcpy(_data + HEAD_ID_LENGTH, &host_max_length, HEAD_DATA_LENGTH);
33
    memcpy(_data + HEAD_TOTAL_LENGTH, msg, max_length);
34
  }
35
private:
36
  int _msg_id;
37
};