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

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

客户端发送格式如 10 + 5。

class request {
public:
    int x_;
    int y_;
    char op_;
};

响应 Response

服务器返回格式如 15 0，其中第二个数字表示状态码。

class response {
public:
    int result_;
    int code_;
};

状态码定义如下：

核心难点：解决 TCP 粘包问题

什么是粘包？

TCP 是面向字节流的，发送方发出的多个数据包可能在接收方看来是一个大包，或者一个大包被拆分成多次读取。

发送：请求 1 + 请求 2 + 请求 3 接收：可能变成 请求 1 请求 2 | 请求 3

服务器的处理方法

在服务器端，我们需要维护一个输入缓冲区。每次 read 到的数据都追加到缓冲区中，然后循环尝试调用 Decode。只有当 Decode 返回 true 时，才说明当前缓冲区里有一条完整的有效消息，可以提取出来处理。

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;
}

服务器核心代码

这里使用 fork 多进程模型处理并发连接。子进程负责读写，父进程继续监听新连接。

while (true) {
    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);
}

可以看到，服务器收到数据后，首先调用 Calculator 方法。如果解码失败（返回空），说明报文不完整，不会立即返回，而是继续接收新数据直到凑齐一条完整报文。这样就保证了每次处理的都是正确格式的独立消息。

完整代码

自定义协议头文件

#include <string>
#define blank_sep " "
#define protocol_sep "\n"

class request {
public:
    request(int x, int y, char op) : x_(x), y_(y), op_(op) {}
    request() {}
    ~request() {}

    std::string serialization() {
        std::string str;
        str += std::to_string(x_);
        str += blank_sep;
        str += op_;
        str += blank_sep;
        str += std::to_string(y_);
        return str;
    }

    bool Deserialization(std::string &in) {
        size_t leftpos = in.find(blank_sep);
        if (leftpos == std::string::npos) return false;
        std::string str_x = in.substr(0, leftpos);
        x_ = std::stoi(str_x);
        op_ = in[leftpos + 1];
        
        size_t rightpos = in.rfind(blank_sep);
        if (rightpos == std::string::npos) return false;
        std::string str_y = in.substr(rightpos + 1);
        y_ = std::stoi(str_y);
        return true;
    }

    void DebugPrint() {
        std::cout << "新请求构建完成：" << x_ << op_ << y_ << "=?" << std::endl;
    }

public:
    int x_;
    int y_;
    char op_;
};

class response {
public:
    response() {}
    response(int result, int code) : result_(result), code_(code) {}
    ~response() {}

    std::string serialization() {
        std::string str;
        str += std::to_string(result_);
        str += blank_sep;
        str += std::to_string(code_);
        return str;
    }

    bool Deserialization(std::string &in) {
        size_t pos = in.find(blank_sep);
        if (pos == std::string::npos) return false;
        std::string str_result = in.substr(0, pos);
        result_ = std::stoi(str_result);
        std::string str_code = in.substr(pos + 1);
        code_ = std::stoi(str_code);
        return true;
    }

    void DebugPrint() {
        std::cout << "结果响应完成，result: " << result_ << ", code: " << code_ << std::endl;
    }

public:
    int result_;
    int code_;
};

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

bool Decode(std::string &s, std::string *content) {
    size_t left_pos = s.find(protocol_sep);
    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;
}

服务端实现

#include <iostream>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <string>
#include <functional>

enum error { SocketErr = 2, BindErr, ListenErr, ConnectErr };

class Socket {
public:
    Socket() {
        sockfd_ = socket(AF_INET, SOCK_STREAM, 0);
        if (sockfd_ < 0) {
            std::cout << "socket fail" << std::endl;
            exit(SocketErr);
        }
    }

    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);
        if (bind(sockfd_, (struct sockaddr *)&server, sizeof(server)) < 0) {
            std::cout << "server bind fail!" << std::endl;
            exit(BindErr);
        }
        std::cout << "server bind successful" << std::endl;
    }

    void Listen() {
        if (listen(sockfd_, 10) < 0) {
            std::cout << "server listen fail!" << std::endl;
            exit(ListenErr);
        }
        std::cout << "server listen successful" << std::endl;
    }

    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);
        if (sockfd < 0) {
            std::cout << "accept fail!" << std::endl;
            return -1;
        }
        std::cout << "accept successful" << std::endl;
        *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 Connect(uint16_t &server_port, std::string &server_ip) {
        struct sockaddr_in server;
        server.sin_family = AF_INET;
        server.sin_port = htons(server_port);
        inet_pton(AF_INET, server_ip.c_str(), &server.sin_addr);
        if (connect(sockfd_, (struct sockaddr *)&server, sizeof server) < 0) {
            std::cout << "connect fail!" << std::endl;
            exit(ConnectErr);
        }
        std::cout << "connect successful!" << std::endl;
    }

    void Close() { close(sockfd_); }
    int fd() { return sockfd_; }
    ~Socket() { close(sockfd_); }

private:
    int sockfd_;
};

class CalculatorServer {
public:
    CalculatorServer() {}

    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;
    }

    ~CalculatorServer() {}
};

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

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();
        std::cout << "init server successful!" << std::endl;
    }

    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);
                            std::cout << info << std::endl;
                            if (info.empty()) { break; }
                            write(sockfd, info.c_str(), info.size());
                        }
                    } else if (s == 0) { break; }
                    else { break; }
                }
                close(sockfd);
                exit(0);
            }
            close(sockfd);
        }
    }

    ~TcpServer() {}

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 <cassert>
#include <unistd.h>
#include "Protocol.hpp"
#include "Socket.hpp"

static void Usage(const std::string &proc) {
    std::cout << "\nUsage: " << 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) {
        std::cout << "===============第" << cnt << "次测试....., ================" << std::endl;
        int x = rand() % 100 + 1;
        usleep(1234);
        int y = rand() % 100;
        usleep(4321);
        char oper = opers[rand() % opers.size()];
        
        request req(x, y, oper);
        req.DebugPrint();
        
        std::string package;
        package = req.serialization();
        package = Encode(package);
        std::cout << package << std::endl;
        
        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);
            assert(r);
            response resp;
            r = resp.Deserialization(content);
            assert(r);
            resp.DebugPrint();
        }
        std::cout << "=================================================" << std::endl;
        sleep(1);
        cnt++;
    }
    sockfd.Close();
    return 0;
}

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