TCP多客户端服务器最佳实现及收发线程方案咨询

你的目标服务端仅需承载4~5个客户端，当前采用的「每连接分配独立接收、发送线程」的架构完全适配需求，不存在选型问题，无需为了追求高性能强行切换复杂度更高的IO多路复用模型，该模型下代码逻辑简单直观，对首次接触网络编程的开发者非常友好，只要修正现有逻辑的缺陷即可稳定运行。

现有代码的核心缺陷 #

• CPU空转问题：发送线程判断队列空时直接执行continue，会导致线程持续占用CPU跑满核心，必须改用阻塞唤醒机制。
• socket写入竞态问题：接收线程处理业务时直接调用send()返回响应，和发送线程的写操作同时作用于同一个socket，TCP是无边界字节流，并发写入会出现数据包拼接错乱，所有socket写操作必须全部收口到发送线程。
• 发送流程阻塞问题：在发送线程中同步阻塞等待ACK，会导致整个发送队列的后续消息全部卡住，若对端异常未返回ACK，发送逻辑会直接死等。
• 心跳逻辑失效问题：现有逻辑仅判断alive_flag阈值，没有实现计数的定时递增逻辑，存活检测无法正常触发。
• 资源释放竞态问题：断连时直接关闭socket，没有同步等待收发线程退出，可能出现线程访问已关闭socket触发崩溃、资源泄漏的问题。
• 共享变量线程安全问题：发送队列、停止标志、ACK状态映射等跨线程访问的资源没有加锁保护，会出现偶现的数据错乱、内存踩踏问题。

适配当前场景的最佳实现方案 #

• 保留现有「1个监听线程 + N对收/发线程（N为客户端数）」的线程模型，4~5个客户端对应的工作线程总数仅10个左右，系统资源开销可以忽略，稳定性远高于强行引入的复杂高性能模型。
• 严格执行socket操作线程归属：每个连接的socket读操作仅允许接收线程调用，写操作仅允许发送线程调用，彻底杜绝跨线程操作socket的竞态。
• 发送队列基于「互斥锁+条件变量」实现阻塞队列：有数据入队时唤醒发送线程，无数据时发送线程挂起阻塞，不占用CPU资源。
• 明确应用层包协议：采用固定包头（携带包总长度、操作码、消息序列号）+ 可变包体的结构，接收时严格按照包头标注的长度收全完整包再处理，保留你现有循环调用send/recv直到整包传输完成的逻辑，同时补充send/recv返回0（对端断开）、返回负值（连接错误/超时/被信号打断）的分支处理，出错直接走统一断连流程。
• 心跳检测合并到接收逻辑：给socket设置5秒接收超时，每次超时未收到数据就给存活计数加1，计数达到5（即25秒无任何数据流入）就触发断连；收到任意客户端数据（包括心跳包、业务包）都直接将存活计数清零，无需单独判断心跳包类型，简化逻辑。
• 统一断连流程收口：不管是接收出错、发送出错、心跳超时触发的断连，都按顺序执行：置连接停止标志→唤醒阻塞的发送线程→等待收发两个工作线程完全退出→关闭socket→清理队列、状态映射等连接相关资源，避免重复释放或资源泄漏。
• 耗时操作隔离：如果业务逻辑存在长耗时操作（如慢数据库查询、大文件读写），不要直接在接收线程里执行，单独部署业务线程池处理任务，处理完成后再将响应消息投递到对应连接的发送队列，避免卡住收发线程导致心跳超时断连。

修正后的核心逻辑参考 #

// 阻塞发送队列实现
class BlockSendQueue {
private:
    std::mutex mtx;
    std::condition_variable cv;
    std::queue<Msg> q;
    bool stop_flag = false;
public:
    void push(Msg msg) {
        std::lock_guard<std::mutex> lk(mtx);
        q.push(std::move(msg));
        cv.notify_one();
    }
    bool pop(Msg& out_msg) {
        std::unique_lock<std::mutex> lk(mtx);
        cv.wait(lk, [&](){ return !q.empty() || stop_flag; });
        if (stop_flag) return false;
        out_msg = std::move(q.front());
        q.pop();
        return true;
    }
    void stop() {
        std::lock_guard<std::mutex> lk(mtx);
        stop_flag = true;
        cv.notify_all();
    }
};

// 发送线程逻辑
void send_thread_routine() {
    Msg msg;
    while(send_queue.pop(msg)) { // 队列为空时阻塞，队列停止时直接退出
        int total_sent = 0;
        // 循环发送直到整包发完
        while(total_sent < msg.total_len) {
            int ret = send(sockfd, msg.buf + total_sent, msg.total_len - total_sent, 0);
            if (ret <= 0) { // 发送出错/连接断开
                trigger_disconnect();
                return;
            }
            total_sent += ret;
        }
        // 不在这里同步等ACK，仅记录待确认的消息序列号，ACK由接收线程统一处理
        if (msg.need_ack) {
            std::lock_guard<std::mutex> lk(ack_map_mtx);
            wait_ack_map[msg.seq] = msg;
        }
    }
}

// 接收线程逻辑
void recv_thread_routine() {
    char recv_buf[8192];
    int alive_cnt = 0;
    // 设置5秒接收超时
    struct timeval tv = {5, 0};
    setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));

    while(!conn_stop_flag) {
        int ret = recv(sockfd, recv_buf, sizeof(recv_buf), 0);
        if (ret < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) { // 接收超时
                alive_cnt++;
                if (alive_cnt >= 5) { // 累计25秒无数据，触发断连
                    trigger_disconnect();
                    return;
                }
                continue;
            }
            // 其余错误直接断连
            trigger_disconnect();
            return;
        }
        if (ret == 0) { // 对端主动关闭连接
            trigger_disconnect();
            return;
        }
        // 收到任意数据，重置心跳计数
        alive_cnt = 0;
        // 将收到的数据追加到解析缓冲区，拆出完整的应用层包
        parse_buffer.append(recv_buf, ret);
        while(parse_buffer.has_full_packet()) {
            Msg full_pack = parse_buffer.pop_packet();
            switch(full_pack.opcode) {
                case OP_ACK: {
                    // 处理ACK，移除待确认映射
                    std::lock_guard<std::mutex> lk(ack_map_mtx);
                    wait_ack_map.erase(full_pack.ack_seq);
                    break;
                }
                case OP_ALIVE:
                case OP_BUSI_A:
                case OP_BUSI_B: {
                    // 处理业务逻辑，响应直接投递到发送队列，不直接调用send
                    Msg resp = build_response(full_pack);
                    send_queue.push(resp);
                    break;
                }
            }
        }
    }
}

内容的提问来源于stack exchange，提问作者mystes