简单echo服务器 -- v2.0

在原有双栈兼容基础上，新增线程池替代进程、文件日志、超时处理、自定义协议四大核心优化，解决进程开销高、排查难、资源占用、粘包等问题，适用于高并发场景。

一、核心优化方案设计

优化功能	实现思路
线程池优化	设计固定大小线程池（基于pthread），复用线程处理客户端连接，避免频繁创建销毁进程的开销
文件日志系统	实现线程安全的日志类，记录时间戳、日志级别、事件详情，写入本地文件（如echo_server.log）
超时处理	通过setsockopt设置SO_RCVTIMEO/SO_SNDTIMEO，为recv/send设置超时（默认 5 秒）
自定义协议	定义 “4 字节长度头 + 数据” 格式，解决 TCP 粘包问题（长度头用网络字节序传输）

二、通用工具类实现（日志 + 线程池）

1. 线程安全的文件日志类（Logger）

负责将日志写入文件，支持INFO/ERROR级别，包含时间戳，多线程下通过互斥锁保证安全。

#include <iostream>
#include <fstream>
#include <time.h>
#include <pthread.h>
#include <cstdarg>
#include <cstring>

// 日志级别枚举
enum LogLevel {
    LOG_INFO,
    LOG_ERROR
};

class Logger {
public:
    // 单例模式（避免多线程重复创建文件句柄）
    static Logger& getInstance() {
        static Logger instance;
        return instance;
    }

    // 禁止拷贝构造和赋值
    Logger(const Logger&) = delete;
    Logger& operator=(const Logger&) = delete;

    // 写入日志（支持可变参数，类似printf）
    void log(LogLevel level, const char* format, ...) {
        pthread_mutex_lock(&logMutex);  // 加锁保证线程安全

        // 1. 获取当前时间戳
        time_t now = time(nullptr);
        struct tm* tmInfo = localtime(&now);
        char timeBuf[32] = {0};
        strftime(timeBuf, sizeof(timeBuf), "%Y-%m-%d %H:%M:%S", tmInfo);

        // 2. 确定日志级别字符串
        const char* levelStr = (level == LOG_INFO) ? "[INFO]" : "[ERROR]";

        // 3. 处理可变参数（格式化日志内容）
        char contentBuf[1024] = {0};
        va_list args;
        va_start(args, format);
        vsnprintf(contentBuf, sizeof(contentBuf)-1, format, args);
        va_end(args);

        // 4. 写入文件（同时输出到控制台）
        logFile << timeBuf << " " << levelStr << " " << contentBuf << std::endl;
        logFile.flush();  // 强制刷新缓冲区，避免日志丢失
        std::cout << timeBuf << " " << levelStr << " " << contentBuf << std::endl;

        pthread_mutex_unlock(&logMutex);  // 解锁
    }

private:
    Logger() {
        // 初始化互斥锁
        pthread_mutex_init(&logMutex, nullptr);
        // 打开日志文件（追加模式，不存在则创建）
        logFile.open("echo_server.log", std::ios::app | std::ios::out);
        if (!logFile.is_open()) {
            std::cerr << "日志文件打开失败！" << std::endl;
        }
    }

    ~Logger() {
        // 释放资源
        if (logFile.is_open()) {
            logFile.close();
        }
        pthread_mutex_destroy(&logMutex);
    }

    std::ofstream logFile;       // 日志文件流
    pthread_mutex_t logMutex;    // 日志写入互斥锁
};

// 全局日志宏（简化调用）
#define LOG_INFO(format, ...)  Logger::getInstance().log(LOG_INFO, format, ##__VA_ARGS__)
#define LOG_ERROR(format, ...) Logger::getInstance().log(LOG_ERROR, format, ##__VA_ARGS__)

2. 固定大小线程池类（ThreadPool）

管理一组线程，循环从任务队列中获取任务执行，支持添加客户端连接处理任务。

#include <queue>
#include <vector>
#include <pthread.h>
#include <sys/socket.h>
#include <arpa/inet.h>

// 任务结构体（存储客户端连接信息）
struct Task {
    int connfd;                      // 客户端连接套接字
    struct sockaddr_storage clientAddr;  // 客户端地址（兼容IPv4/IPv6）
};

class ThreadPool {
public:
    // 初始化线程池（参数：线程数量）
    ThreadPool(int threadNum) : threadCount(threadNum), isRunning(false) {
        // 初始化互斥锁和条件变量
        pthread_mutex_init(&queueMutex, nullptr);
        pthread_cond_init(&queueCond, nullptr);
    }

    ~ThreadPool() {
        stop();  // 停止线程池
        // 释放资源
        pthread_mutex_destroy(&queueMutex);
        pthread_cond_destroy(&queueCond);
    }

    // 启动线程池
    void start() {
        if (isRunning) return;
        isRunning = true;
        // 创建指定数量的线程
        threads.resize(threadCount);
        for (int i = 0; i < threadCount; ++i) {
            if (pthread_create(&threads[i], nullptr, threadFunc, this) != 0) {
                LOG_ERROR("创建线程%d失败：%s", i, strerror(errno));
                isRunning = false;
                break;
            }
        }
        if (isRunning) {
            LOG_INFO("线程池启动成功，线程数量：%d", threadCount);
        }
    }

    // 停止线程池（等待所有线程退出）
    void stop() {
        if (!isRunning) return;
        isRunning = false;
        pthread_cond_broadcast(&queueCond);  // 唤醒所有等待的线程

        // 等待所有线程退出
        for (pthread_t& tid : threads) {
            pthread_join(tid, nullptr);
        }
        threads.clear();
        LOG_INFO("线程池已停止");
    }

    // 添加任务到队列（外部调用，如accept后添加客户端任务）
    bool addTask(const Task& task) {
        pthread_mutex_lock(&queueMutex);
        // 任务队列满（简单限制队列最大长度为1024）
        if (taskQueue.size() >= 1024) {
            pthread_mutex_unlock(&queueMutex);
            LOG_ERROR("任务队列已满，拒绝新连接");
            return false;
        }
        taskQueue.push(task);
        pthread_mutex_unlock(&queueMutex);
        pthread_cond_signal(&queueCond);  // 唤醒一个等待的线程
        return true;
    }

private:
    // 线程入口函数（静态函数，通过this指针访问成员）
    static void* threadFunc(void* arg) {
        ThreadPool* pool = static_cast<ThreadPool*>(arg);
        pool->run();  // 线程循环执行任务
        return nullptr;
    }

    // 线程循环逻辑（取任务、执行任务）
    void run() {
        while (isRunning) {
            Task task;
            pthread_mutex_lock(&queueMutex);

            // 等待任务（队列为空且线程池运行中）
            while (isRunning && taskQueue.empty()) {
                pthread_cond_wait(&queueCond, &queueMutex);
            }

            // 线程池已停止，退出循环
            if (!isRunning) {
                pthread_mutex_unlock(&queueMutex);
                break;
            }

            // 取出任务
            task = taskQueue.front();
            taskQueue.pop();
            pthread_mutex_unlock(&queueMutex);

            // 执行任务（处理客户端回声请求）
            handleClientTask(task);
        }
    }

    // 处理单个客户端任务（核心逻辑，替代原fork子进程）
    void handleClientTask(const Task& task) {
        int connfd = task.connfd;
        struct sockaddr_storage clientAddr = task.clientAddr;

        // 1. 打印并记录客户端IP
        char ipstr[INET6_ADDRSTRLEN] = {0};
        void* addr = (clientAddr.ss_family == AF_INET) ? 
            &((struct sockaddr_in*)&clientAddr)->sin_addr : 
            &((struct sockaddr_in6*)&clientAddr)->sin6_addr;
        inet_ntop(clientAddr.ss_family, addr, ipstr, sizeof(ipstr));
        LOG_INFO("新客户端连接：%s，connfd：%d", ipstr, connfd);

        // 2. 设置套接字超时（recv/send超时5秒）
        struct timeval timeout = {5, 0};  // 5秒超时
        // 设置接收超时
        if (setsockopt(connfd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) == -1) {
            LOG_ERROR("connfd=%d 设置接收超时失败：%s", connfd, strerror(errno));
            close(connfd);
            return;
        }
        // 设置发送超时
        if (setsockopt(connfd, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout)) == -1) {
            LOG_ERROR("connfd=%d 设置发送超时失败：%s", connfd, strerror(errno));
            close(connfd);
            return;
        }

        // 3. 处理回声请求（基于自定义协议）
        handleEchoWithProtocol(connfd, ipstr);

        // 4. 释放连接资源
        close(connfd);
        LOG_INFO("客户端%s 连接关闭，connfd：%d", ipstr, connfd);
    }

    // 基于自定义协议的回声处理（解决粘包）
    void handleEchoWithProtocol(int connfd, const char* clientIP) {
        char recvBuf[1024] = {0};    // 接收缓冲区
        char sendBuf[1024] = {0};    // 发送缓冲区
        ssize_t n;

        while (true) {
            // -------------------------- 步骤1：接收协议头（4字节长度） --------------------------
            uint32_t dataLen = 0;  // 存储数据长度（网络字节序转主机字节序后）
            // 接收4字节长度头（循环接收，确保完整）
            n = recvN(connfd, (char*)&dataLen, sizeof(dataLen));
            if (n == -1) {
                if (errno == EAGAIN || errno == EWOULDBLOCK) {
                    LOG_ERROR("connfd=%d 接收超时（客户端%s无数据发送）", connfd, clientIP);
                } else {
                    LOG_ERROR("connfd=%d 接收长度头失败：%s", connfd, strerror(errno));
                }
                break;
            } else if (n == 0) {
                LOG_INFO("客户端%s 主动关闭连接，connfd：%d", clientIP, connfd);
                break;
            }

            // 网络字节序转主机字节序（大端转小端）
            dataLen = ntohl(dataLen);
            if (dataLen > sizeof(recvBuf)) {  // 限制最大数据长度，避免缓冲区溢出
                LOG_ERROR("connfd=%d 数据长度超出限制（%d > %lu）", connfd, dataLen, sizeof(recvBuf));
                break;
            }

            // -------------------------- 步骤2：接收数据内容 --------------------------
            n = recvN(connfd, recvBuf, dataLen);
            if (n == -1) {
                LOG_ERROR("connfd=%d 接收数据失败：%s", connfd, strerror(errno));
                break;
            } else if (n == 0) {
                LOG_INFO("客户端%s 主动关闭连接，connfd：%d", clientIP, connfd);
                break;
            }

            // 记录接收的数据
            LOG_INFO("connfd=%d 接收客户端%s 数据：%s（长度：%d）", connfd, clientIP, recvBuf, dataLen);

            // -------------------------- 步骤3：发送回声数据（按协议打包） --------------------------
            // 数据内容复制到发送缓冲区
            strncpy(sendBuf, recvBuf, dataLen);
            // 打包协议头（主机字节序转网络字节序）
            uint32_t sendLen = htonl(dataLen);
            // 先发送长度头
            if (sendN(connfd, (char*)&sendLen, sizeof(sendLen)) == -1) {
                LOG_ERROR("connfd=%d 发送长度头失败：%s", connfd, strerror(errno));
                break;
            }
            // 再发送数据内容
            if (sendN(connfd, sendBuf, dataLen) == -1) {
                LOG_ERROR("connfd=%d 发送数据失败：%s", connfd, strerror(errno));
                break;
            }

            LOG_INFO("connfd=%d 回声客户端%s 数据：%s（长度：%d）", connfd, clientIP, sendBuf, dataLen);
            memset(recvBuf, 0, sizeof(recvBuf));  // 清空缓冲区
        }
    }

    // 封装recv：确保接收指定长度的数据（解决部分接收问题）
    ssize_t recvN(int fd, char* buf, size_t len) {
        size_t total = 0;
        while (total < len) {
            ssize_t n = recv(fd, buf + total, len - total, 0);
            if (n == -1) {
                return -1;  // 错误（超时或其他错误）
            } else if (n == 0) {
                return total;  // 客户端关闭，返回已接收长度
            }
            total += n;
        }
        return total;  // 接收完成，返回总长度
    }

    // 封装send：确保发送指定长度的数据（解决部分发送问题）
    ssize_t sendN(int fd, const char* buf, size_t len) {
        size_t total = 0;
        while (total < len) {
            ssize_t n = send(fd, buf + total, len - total, 0);
            if (n == -1) {
                return -1;  // 错误（超时或其他错误）
            }
            total += n;
        }
        return total;  // 发送完成，返回总长度
    }

private:
    std::vector<pthread_t> threads;    // 线程列表
    std::queue<Task> taskQueue;        // 任务队列
    pthread_mutex_t queueMutex;        // 任务队列互斥锁
    pthread_cond_t queueCond;          // 任务队列条件变量
    int threadCount;                   // 线程数量
    bool isRunning;                    // 线程池运行状态
};

三、增强版服务器实现

基于原有双栈服务器，替换fork为线程池，集成日志、超时、自定义协议。

#include <sys/socket.h>
#include <sys/types.h>
#include <netdb.h>
#include <cstring>
#include <unistd.h>
#include <arpa/inet.h>
#include <cerrno>
#include <cstdlib>

// 包含上文的Logger和ThreadPool类

// 创建双栈监听套接字（复用原有逻辑，添加日志）
int createListener(const char* service = "9527") {
    struct addrinfo hints, *result, *p;
    memset(&hints, 0, sizeof(hints));
    hints.ai_family = AF_UNSPEC;       // 双栈兼容
    hints.ai_socktype = SOCK_STREAM;   // TCP类型
    hints.ai_flags = AI_PASSIVE;       // 通配地址绑定

    int err = getaddrinfo(NULL, service, &hints, &result);
    if (err) {
        LOG_ERROR("getaddrinfo解析失败：%s", gai_strerror(err));
        return -1;
    }

    int listenfd = -1;
    for (p = result; p != nullptr; p = p->ai_next) {
        listenfd = socket(p->ai_family, p->ai_socktype, p->ai_protocol);
        if (listenfd == -1) {
            LOG_ERROR("创建套接字失败：%s（尝试下一个地址）", strerror(errno));
            continue;
        }

        // 设置地址重用（避免重启端口占用）
        int opt = 1;
        if (setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == -1) {
            LOG_ERROR("setsockopt失败：%s", strerror(errno));
            close(listenfd);
            continue;
        }

        if (bind(listenfd, p->ai_addr, p->ai_addrlen) == -1) {
            LOG_ERROR("绑定端口失败：%s（尝试下一个地址）", strerror(errno));
            close(listenfd);
            continue;
        }

        if (listen(listenfd, 10) == -1) {  // backlog设为10，支持更多等待连接
            LOG_ERROR("监听失败：%s", strerror(errno));
            close(listenfd);
            continue;
        }

        break;  // 成功创建监听套接字
    }

    freeaddrinfo(result);
    if (p == nullptr || listenfd == -1) {
        LOG_ERROR("所有地址尝试失败，无法创建监听套接字");
        return -1;
    }

    LOG_INFO("双栈服务器启动成功，监听端口：%s（支持IPv4/IPv6）", service);
    return listenfd;
}

int main(int argc, char* argv[]) {
    // 1. 初始化线程池（线程数量：8，可根据CPU核心数调整）
    ThreadPool threadPool(8);
    threadPool.start();

    // 2. 创建双栈监听套接字（默认端口9527，支持命令行指定：./server 8888）
    const char* service = (argc > 1) ? argv[1] : "9527";
    int listenfd = createListener(service);
    if (listenfd == -1) {
        threadPool.stop();
        return 1;
    }

    // 3. 循环接受客户端连接，添加到线程池任务队列
    while (true) {
        struct sockaddr_storage clientAddr;
        socklen_t clientAddrLen = sizeof(clientAddr);
        int connfd = accept(listenfd, (struct sockaddr*)&clientAddr, &clientAddrLen);

        if (connfd == -1) {
            LOG_ERROR("accept失败：%s", strerror(errno));
            continue;
        }

        // 封装任务，添加到线程池
        Task task = {connfd, clientAddr};
        if (!threadPool.addTask(task)) {
            close(connfd);  // 任务队列满，关闭连接
            LOG_ERROR("任务队列满，拒绝客户端连接（connfd：%d）", connfd);
        }
    }

    // 理论上不会执行到这里（无限循环）
    close(listenfd);
    threadPool.stop();
    return 0;
}

四、增强版客户端实现

客户端需按 “4 字节长度头 + 数据” 格式发送 / 接收数据，适配服务器的协议优化。

#include <iostream>
#include <sys/socket.h>
#include <sys/types.h>
#include <netdb.h>
#include <cstring>
#include <unistd.h>
#include <arpa/inet.h>
#include <cerrno>
#include <cstdlib>
#include <cstdio>

// 封装sendN：确保发送指定长度（适配自定义协议）
ssize_t sendN(int fd, const char* buf, size_t len) {
    size_t total = 0;
    while (total < len) {
        ssize_t n = send(fd, buf + total, len - total, 0);
        if (n == -1) {
            std::cerr << "发送失败：" << strerror(errno) << std::endl;
            return -1;
        }
        total += n;
    }
    return total;
}

// 封装recvN：确保接收指定长度（适配自定义协议）
ssize_t recvN(int fd, char* buf, size_t len) {
    size_t total = 0;
    while (total < len) {
        ssize_t n = recv(fd, buf + total, len - total, 0);
        if (n == -1) {
            std::cerr << "接收失败：" << strerror(errno) << std::endl;
            return -1;
        } else if (n == 0) {
            std::cerr << "服务器已关闭连接" << std::endl;
            return total;
        }
        total += n;
    }
    return total;
}

// 建立双栈连接（复用原有逻辑）
int connectToServer(const char* node = "127.0.0.1", const char* service = "9527") {
    struct addrinfo hints, *result, *p;
    memset(&hints, 0, sizeof(hints));
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;

    int err = getaddrinfo(node, service, &hints, &result);
    if (err) {
        std::cerr << "解析地址失败：" << gai_strerror(err) << std::endl;
        return -1;
    }

    int connfd = -1;
    for (p = result; p != nullptr; p = p->ai_next) {
        connfd = socket(p->ai_family, p->ai_socktype, p->ai_protocol);
        if (connfd == -1) {
            std::cerr << "创建套接字失败：" << strerror(errno) << "（尝试下一个地址）" << std::endl;
            continue;
        }

        if (connect(connfd, p->ai_addr, p->ai_addrlen) == -1) {
            std::cerr << "连接失败：" << strerror(errno) << "（尝试下一个地址）" << std::endl;
            close(connfd);
            continue;
        }

        break;
    }

    freeaddrinfo(result);
    if (p == nullptr || connfd == -1) {
        std::cerr << "无法连接到服务器 " << node << ":" << service << std::endl;
        return -1;
    }

    std::cout << "成功连接到 " << node << ":" << service << std::endl;
    return connfd;
}

// 基于自定义协议的交互逻辑
void handleEchoInteraction(int connfd) {
    char inputBuf[1024] = {0};
    char recvBuf[1024] = {0};

    std::cout << "请输入要发送的内容（输入quit退出）：" << std::endl;
    while (true) {
        // 1. 读取用户输入
        if (!fgets(inputBuf, sizeof(inputBuf), stdin)) {
            std::cerr << "\n输入错误" << std::endl;
            break;
        }

        // 2. 处理输入（去除换行符，检查退出）
        size_t inputLen = strlen(inputBuf);
        if (inputLen > 0 && inputBuf[inputLen - 1] == '\n') {
            inputBuf[--inputLen] = '\0';  // 去除换行符
        }
        if (strcmp(inputBuf, "quit") == 0) {
            std::cout << "正在退出..." << std::endl;
            break;
        }

        // 3. 按自定义协议打包发送（长度头+数据）
        uint32_t sendLen = htonl(inputLen);  // 主机字节序转网络字节序
        // 先发送4字节长度头
        if (sendN(connfd, (char*)&sendLen, sizeof(sendLen)) == -1) {
            close(connfd);
            return;
        }
        // 再发送数据内容
        if (sendN(connfd, inputBuf, inputLen) == -1) {
            close(connfd);
            return;
        }
        std::cout << "已发送：" << inputBuf << "（长度：" << inputLen << "）" << std::endl;

        // 4. 按自定义协议接收响应（长度头+数据）
        uint32_t recvLen = 0;
        // 先接收4字节长度头
        if (recvN(connfd, (char*)&recvLen, sizeof(recvLen)) == -1) {
            close(connfd);
            return;
        }
        recvLen = ntohl(recvLen);  // 网络字节序转主机字节序
        // 再接收数据内容
        if (recvN(connfd, recvBuf, recvLen) == -1) {
            close(connfd);
            return;
        }
        recvBuf[recvLen] = '\0';  // 确保字符串终止

        // 5. 显示响应
        std::cout << "服务器响应：" << recvBuf << "（长度：" << recvLen << "）" << std::endl;
        memset(inputBuf, 0, sizeof(inputBuf));
        memset(recvBuf, 0, sizeof(recvBuf));
    }

    close(connfd);
    std::cout << "已断开连接" << std::endl;
}

int main(int argc, char* argv[]) {
    // 支持命令行参数：./client 服务器IP 端口（默认127.0.0.1:9527）
    const char* node = (argc > 1) ? argv[1] : "127.0.0.1";
    const char* service = (argc > 2) ? argv[2] : "9527";

    int connfd = connectToServer(node, service);
    if (connfd == -1) {
        return 1;
    }

    handleEchoInteraction(connfd);
    return 0;
}

五、编译与使用说明

1. 编译命令（Linux 环境）

需链接pthread库（线程相关），将所有代码合并编译（或分文件编译）：

# 服务器编译（包含Logger、ThreadPool、服务器逻辑）
g++ -o echo_server enhanced_server.cpp -std=c++11 -pthread
# 客户端编译
g++ -o echo_client enhanced_client.cpp -std=c++11

2. 运行步骤

步骤 1：启动服务器

./echo_server 9527  # 监听9527端口（默认）
# 日志输出示例：
# 2025-09-12 15:30:00 [INFO] 线程池启动成功，线程数量：8
# 2025-09-12 15:30:00 [INFO] 双栈服务器启动成功，监听端口：9527（支持IPv4/IPv6）

步骤 2：启动客户端（可多开）

# 连接IPv4服务器
./echo_client 127.0.0.1 9527
# 连接IPv6服务器
./echo_client ::1 9527

步骤 3：测试交互

# 客户端输入
请输入要发送的内容（输入quit退出）：
hello enhanced tcp
已发送：hello enhanced tcp（长度：18）
服务器响应：hello enhanced tcp（长度：18）
quit
正在退出...
已断开连接

步骤 4：查看日志文件

服务器运行时会生成echo_server.log，包含所有事件记录：

2025-09-12 15:30:10 [INFO] 新客户端连接：127.0.0.1，connfd：4
2025-09-12 15:30:15 [INFO] connfd=4 接收客户端127.0.0.1 数据：hello enhanced tcp（长度：18）
2025-09-12 15:30:15 [INFO] connfd=4 回声客户端127.0.0.1 数据：hello enhanced tcp（长度：18）
2025-09-12 15:30:20 [INFO] 客户端127.0.0.1 主动关闭连接，connfd：4
2025-09-12 15:30:20 [INFO] 客户端127.0.0.1 连接关闭，connfd：4

六、优化效果验证

线程池优化：通过ps -T -p <服务器PID>查看线程数量，仅启动 8 个线程（原 fork 会创建大量子进程），CPU 和内存开销显著降低。

日志系统：echo_server.log记录所有关键事件，便于排查连接失败、超时等问题。

超时处理：客户端 5 秒不发送数据，服务器会自动关闭连接，避免线程长期阻塞。

粘包解决：连续发送多条短数据（如 “a”“b”），服务器会按协议拆分为两条独立数据，无粘包现象。