手写 C++ TCP 服务器：自定义协议与粘包处理实战

TCP 网络编程中的粘包问题与自定义协议设计

在网络编程中，UDP 基于数据报传输，天然具备消息边界。但 TCP 不同，它是字节流协议，没有明确的消息界限。这意味着发送方写入的数据，接收方可能一次读取一部分，也可能多次读取才凑齐一条完整消息。

为什么需要应用层协议？

TCP 通信时，客户端通过 connect 建立连接后，使用 write 将数据写入 socket 文件描述符。内核负责何时发送，而应用层只负责写入。这导致接收端无法直接知道一条消息的结束位置。

想象一下，如果发送方先发了一个整数，接着发一个浮点数，而接收方按字符串去解析，数据就会错位。因此，一旦涉及结构化数据，必须制定应用层协议来界定消息边界。

TCP 主要负责控制发送时机、数据量及错误处理，但具体的业务数据格式（如序列化后的结构）需由我们定义。否则，接收端面对连续的字节流，就像收到一段断断续续的语音，难以还原原意。

自定义应用层协议设计

协议格式

为了解决粘包和半包问题，我们采用'长度 + 内容'的定长头部方案：

len\n
content\n

例如，发送 10 + 5，实际传输的是：

6\n10 + 5\n

这种格式让接收方能先读取长度，再精确截取对应字节的内容。

协议封装实现

Encode(封装报文)

std::string Encode(std::string &content) {
    std::string s;
    size_t len = content.size();
    s += std::to_string(len);
    s += "\n";
    s += content;
    s += "\n";
    return s;
}

Decode(解析报文)

解码函数需要完成三件事：判断头部是否完整、校验数据长度、提取内容并移除已读部分。

bool Decode(std::string &s, std::string *content) {
    size_t left_pos = s.find("\n");
    if (left_pos == std::string::npos) return false;
    
    std::string content_len = s.substr(0, left_pos);
    int len = std::stoi(content_len);
    
    // 检查缓冲区是否有足够的数据
    if (s.size() < content_len.size() + len + 2) return false;
    
    *content = s.substr(left_pos + 1, len);
    s.erase(0, content_len.size() + len + 2);
    return true;
}

请求与响应设计

为了演示协议的实际应用，我们构建一个简单的网络计算器。

请求 request

客户端发送格式：x op y，例如 10 + 5。

class request {
public:
    int x_;
    int y_;
    char op_;
    // ... 序列化与反序列化逻辑 ...
};

响应 response

服务器返回格式：result code，例如 15 0。

code	含义
0	成功
1	除 0
2	取模 0
3	非法操作

class response {
public:
    int result_;
    int code_;
    // ... 序列化与反序列化逻辑 ...
};

核心挑战：解决 TCP 粘包问题

什么是粘包？

TCP 是面向字节流的协议。发送方连续发送多个请求，接收方可能一次性读到多个请求，也可能一个请求被拆分成多次读取。

发送：请求 1 + 请求 2

接收：请求 1 请求 2 或 请求 1 + 请求 2

服务器的处理方法

在服务器端，我们需要维护一个输入缓冲区 inbuffer_stream。每次 read 到数据后，追加到缓冲区，然后尝试调用 Decode 解析。

如果 Decode 返回 false，说明数据不完整，继续等待下一次 read。只有当解析成功，才进行业务逻辑处理并发送响应。

while (true) {
    int sockfd = listenfd_.Accept(&client_port, &client_ip);
    if (fork() == 0) {
        while (1) {
            char buffer[1280];
            ssize_t s = read(sockfd, buffer, sizeof buffer - 1);
            if (s > 0) {
                inbuffer_stream += buffer;
                while (true) {
                    std::string info = callback_(inbuffer_stream);
                    if (info.empty()) break;
                    write(sockfd, info.c_str(), info.size());
                }
            } else if (s == 0) {
                break;
            } else {
                break;
            }
        }
        close(sockfd);
        exit(0);
    }
    close(sockfd);
}

这段代码展示了典型的并发模型：主进程监听，子进程处理单个连接。子进程中循环读取数据，利用 callback_ 回调函数处理粘包逻辑，确保每次处理的都是完整报文。

完整代码实现

服务端核心逻辑

#include <iostream>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <string>
#include <functional>

// Socket 封装类
enum error { SocketErr = 2, BindErr, ListenErr, ConnectErr, };
class Socket {
public:
    Socket() { sockfd_ = socket(AF_INET, SOCK_STREAM, 0); }
    void Bind(uint16_t &port, std::string &ip) {
        struct sockaddr_in server;
        server.sin_family = AF_INET;
        server.sin_port = htons(port);
        inet_pton(AF_INET, ip.c_str(), &server.sin_addr);
        bind(sockfd_, (struct sockaddr *)&server, sizeof(server));
    }
    void Listen() { listen(sockfd_, 10); }
    int Accept(uint16_t *client_port, std::string *client_ip) {
        struct sockaddr_in client;
        socklen_t len = sizeof(client);
        int sockfd = accept(sockfd_, (struct sockaddr *)&client, &len);
        *client_port = ntohs(client.sin_port);
        char ip[64];
        inet_ntop(AF_INET, &client.sin_addr, ip, sizeof ip);
        *client_ip = ip;
        return sockfd;
    }
    void Close() { close(sockfd_); }
    int fd() { return sockfd_; }
private:
    int sockfd_;
};

// 计算逻辑服务
class CalculatorServer {
public:
    response CalculatorHandler(const request &req) {
        response res(0, 0);
        switch (req.op_) {
            case '+': res.result_ = req.x_ + req.y_; break;
            case '-': res.result_ = req.x_ - req.y_; break;
            case '*': res.result_ = req.x_ * req.y_; break;
            case '/': if (req.y_ == 0) { res.code_ = 1; break; } res.result_ = req.x_ / req.y_; break;
            case '%': if (req.y_ == 0) { res.code_ = 2; break; } res.result_ = req.x_ % req.y_; break;
            default: res.code_ = 3; break;
        }
        return res;
    }
    std::string Calculator(std::string& s) {
        std::string content;
        if (Decode(s, &content) == false) return "";
        request req;
        bool r = req.Deserialization(content);
        if (!r) return "";
        response res = CalculatorHandler(req);
        std::string ret = res.serialization();
        ret = Encode(ret);
        return ret;
    }
};

using func_t = std::function<std::string(std::string &)>;

// TCP 服务器框架
class TcpServer {
public:
    TcpServer(uint16_t port, std::string ip, func_t callback) : port_(port), ip_(ip), callback_(callback) {}
    void InitServer() {
        listenfd_.Bind(port_, ip_);
        listenfd_.Listen();
    }
    void start() {
        signal(SIGCHLD, SIG_IGN);
        signal(SIGPIPE, SIG_IGN);
        while (true) {
            uint16_t client_port;
            std::string client_ip;
            int sockfd = listenfd_.Accept(&client_port, &client_ip);
            if (sockfd < 0) continue;
            if (fork() == 0) {
                listenfd_.Close();
                std::string inbuffer_stream;
                while (1) {
                    char buffer[1280];
                    ssize_t s = read(sockfd, buffer, sizeof buffer - 1);
                    if (s > 0) {
                        buffer[s] = 0;
                        inbuffer_stream += buffer;
                        while (true) {
                            std::string info = callback_(inbuffer_stream);
                            if (info.empty()) break;
                            write(sockfd, info.c_str(), info.size());
                        }
                    } else if (s == 0) break;
                    else break;
                }
                close(sockfd);
                exit(0);
            }
            close(sockfd);
        }
    }
private:
    Socket listenfd_;
    uint16_t port_;
    std::string ip_;
    func_t callback_;
};

int main(int argc, char* argv[]) {
    if(argc != 3) exit(0);
    uint16_t server_port = std::atoi(argv[2]);
    std::string server_ip = argv[1];
    CalculatorServer cal;
    TcpServer* ser = new TcpServer(server_port, server_ip, std::bind(&CalculatorServer::Calculator, &cal, std::placeholders::_1));
    ser->InitServer();
    ser->start();
    return 0;
}

客户端示例

#include <iostream>
#include <unistd.h>
#include "Protocol.hpp"
#include "Socket.hpp"

static void Usage(const std::string &proc) {
    std::cout << "Usage: " << proc << " serverip serverport\n" << std::endl;
}

int main(int argc, char *argv[]) {
    if (argc != 3) { Usage(argv[0]); exit(0); }
    std::string serverip = argv[1];
    uint16_t serverport = std::stoi(argv[2]);
    Socket sockfd;
    sockfd.Connect(serverport, serverip);
    srand(time(nullptr) ^ getpid());
    int cnt = 1;
    const std::string opers = "+-*/%=-=&^";
    std::string inbuffer_stream;
    while (cnt <= 10) {
        int x = rand() % 100 + 1;
        usleep(1234);
        int y = rand() % 100;
        usleep(4321);
        char oper = opers[rand() % opers.size()];
        request req(x, y, oper);
        std::string package = req.serialization();
        package = Encode(package);
        write(sockfd.fd(), package.c_str(), package.size());
        char buffer[128];
        ssize_t n = read(sockfd.fd(), buffer, sizeof(buffer) - 1);
        if (n > 0) {
            buffer[n] = 0;
            inbuffer_stream += buffer;
            std::string content;
            bool r = Decode(inbuffer_stream, &content);
            response resp;
            r = resp.Deserialization(content);
            resp.DebugPrint();
        }
        sleep(1);
        cnt++;
    }
    sockfd.Close();
    return 0;
}

总结

到这里，我们已经完整实现了一个基于 C++ 的 TCP 计算器服务器。通过这个实践，我们应该建立这样的认知：TCP 编程的本质不是收发数据，而是如何正确解析数据的边界。由于 TCP 没有消息边界，我们必须设计应用层协议来保证数据的完整性。