一、粘包问题基本概念
网络粘包是指在TCP流式传输中，由于TCP协议的特性，多个数据包可能被合并成一个数据包接收（粘包），或者一个数据包被分割成多个数据包接收（半包）。这会导致接收端无法正确识别消息边界，从而造成数据解析错误。
主要表现形式：
- 粘包问题：多个数据包被合并成一个数据包接收
- 拆包问题：单个应用层报文被分割成多个TCP段接收
示例情况：
发送端： [Packet1] [Packet2] [Packet3]
接收端可能收到：
情况1: [Packet1] [Packet2] [Packet3]（理想情况）
情况2: [Packet1] [Packet2部分数据]（半包）
情况3: [Packet1部分] [Packet2] [Packet3部分]（粘包+半包）
二、长度前缀（Length-Prefix）协议
最常用的解决方案是采用”长度+数据”的格式进行通信，即在每个消息前添加一个固定长度的字段，用于表示后续数据的长度。
1
| [4字节长度字段（网络字节序）][实际数据（长度由头部指定）]
|
长度字段使用 uint32_t 大端序（网络字节序）
三、发送端处理
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
| #include <arpa/inet.h>
#include <vector>
void send_message(int sockfd, const std::vector<char>& data) {
uint32_t data_length = static_cast<uint32_t>(data.size());
uint32_t net_length = htonl(data_length);
std::vector<char> buffer(sizeof(net_length) + data.size());
memcpy(buffer.data(), &net_length, sizeof(net_length));
memcpy(buffer.data() + sizeof(net_length), data.data(), data.size());
size_t total_sent = 0;
while (total_sent < buffer.size()) {
ssize_t sent = send(sockfd, buffer.data() + total_sent,
buffer.size() - total_sent, 0);
if (sent == -1) {
if (errno == EINTR) continue;
throw std::runtime_error("send failed: " + std::string(strerror(errno)));
}
total_sent += sent;
}
}
|
四、接收端处理
4.1 基本接收方法
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
|
uint32_t network_length;
size_t total_read = 0;
while (total_read < sizeof(uint32_t)) {
ssize_t n = recv(socket_fd, (char*)&network_length + total_read,
sizeof(uint32_t) - total_read, 0);
if (n <= 0) { }
total_read += n;
}
uint32_t length = ntohl(network_length);
std::vector<char> buffer(length);
size_t total_read_body = 0;
while (total_read_body < length) {
ssize_t n = recv(socket_fd, buffer.data() + total_read_body,
length - total_read_body, 0);
if (n == 0) {
throw std::runtime_error("Connection closed by peer");
} else if (n < 0) {
if (errno == EINTR) continue;
throw std::runtime_error("recv error: " + std::string(strerror(errno)));
}
total_read_body += n;
}
process_message(buffer, length);
|
4.2 处理粘包和半包的完整状态机实现
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
| #include <sys/socket.h>
#include <stdexcept>
class MessageParser {
public:
static constexpr uint32_t MAX_MSG_SIZE = 16 * 1024 * 1024;
enum class ParseState {
ReadingHeader,
ReadingBody
};
std::vector<std::vector<char>> parse(const char* input, size_t len) {
std::vector<std::vector<char>> complete_messages;
size_t processed = 0;
while (processed < len) {
switch (current_state_) {
case ParseState::ReadingHeader: {
size_t remain = HEADER_SIZE - header_bytes_received_;
size_t to_copy = std::min(remain, len - processed);
memcpy(header_buf_ + header_bytes_received_, input + processed, to_copy);
header_bytes_received_ += to_copy;
processed += to_copy;
if (header_bytes_received_ == HEADER_SIZE) {
uint32_t net_length;
memcpy(&net_length, header_buf_, sizeof(net_length));
current_body_length_ = ntohl(net_length);
if (current_body_length_ > MAX_MSG_SIZE) {
throw std::runtime_error("Message size exceeds limit");
}
current_body_.resize(current_body_length_);
body_bytes_received_ = 0;
current_state_ = ParseState::ReadingBody;
}
break;
}
case ParseState::ReadingBody: {
size_t remain = current_body_length_ - body_bytes_received_;
size_t to_copy = std::min(remain, len - processed);
memcpy(current_body_.data() + body_bytes_received_,
input + processed, to_copy);
body_bytes_received_ += to_copy;
processed += to_copy;
if (body_bytes_received_ == current_body_length_) {
complete_messages.push_back(std::move(current_body_));
reset();
}
break;
}
}
}
return complete_messages;
}
private:
void reset() {
current_state_ = ParseState::ReadingHeader;
header_bytes_received_ = 0;
body_bytes_received_ = 0;
current_body_.clear();
}
static constexpr size_t HEADER_SIZE = sizeof(uint32_t);
ParseState current_state_ = ParseState::ReadingHeader;
char header_buf_[HEADER_SIZE] = {0};
size_t header_bytes_received_ = 0;
uint32_t current_body_length_ = 0;
size_t body_bytes_received_ = 0;
std::vector<char> current_body_;
};
|
4.3 完整接收流程
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 message_loop(int sockfd) {
MessageParser parser;
std::vector<char> recv_buf(4096);
while (true) {
ssize_t n = recv(sockfd, recv_buf.data(), recv_buf.size(), 0);
if (n == 0) {
break;
} else if (n < 0) {
if (errno == EINTR) continue;
if (errno == EAGAIN || errno == EWOULDBLOCK) {
usleep(1000);
continue;
}
throw std::runtime_error("recv error: " + std::string(strerror(errno)));
}
auto messages = parser.parse(recv_buf.data(), n);
for (auto& msg : messages) {
process_message(msg.data(), msg.size());
}
}
}
|
五、关键处理要点
- 状态机处理：使用状态机区分正在读取头部还是读取消息体
- 缓冲区管理：动态调整缓冲区大小，根据需要存储部分接收的数据
- 字节累积：持续累积接收的字节，直到获取完整的消息
- 多消息处理：一次接收可能包含多个完整消息，需要全部解析出来
- 安全检查：验证消息长度的合理性，防止恶意攻击
- 字节序问题：在不同架构的计算机之间通信时，必须使用网络字节序（大端序）。使用
htonl()将主机字节序转换为网络字节序，使用ntohl()将网络字节序转换为主机字节序
- 长度字段大小：通常使用4字节（uint32_t）来表示长度，这足以表示最大4GB的消息
- 完整接收：TCP不保证一次
recv调用能接收到完整数据，可能需要循环接收直到获取所需的全部字节
六、总结
TCP流式协议中处理粘包和半包问题的核心是：
- 使用固定长度的头部标识消息长度
- 使用状态机和缓冲区管理接收的数据
- 阻塞等待直到接收到完整数据
- 单次接收可能同时包含多个完整消息和不完整消息，需要综合处理
- 不同的实现方式（memcpy、string操作）各有优缺点，可根据具体需求选择