[计网底层小探索]:实现并部署多线程并发Tcp服务器框架(基于生产者消费者模型的线程池结构)

文章目录

• 一.网络层与传输层协议
• • sockaddr结构体继承体系(Linux体系)
  • 贯穿计算机系统的网络通信架构图示:
  • 二.实现并部署多线程并发Tcp服务器框架
  • • 线程池模块
    • 序列化反序列化工具模块
    • 通信信道建立模块
    • 服务器主体模块
    • 任务回调模块(根据具体应用场景可重构)
    • Tips:DebugC++代码过程中遇到的问题记录

      一.网络层与传输层协议

      • 网络层与传输层内置于操作系统的内核中,网络层一般使用ip协议,传输层常用协议为Tcp协议和Udp协议,Tcp协议和Udp协议拥有各自的特点和应用场景:

        sockaddr结构体继承体系(Linux体系)

        • sockaddr_in结构体用于存储网络通信主机进程的ip和端口号等信息

          贯穿计算机系统的网络通信架构图示:

          二.实现并部署多线程并发Tcp服务器框架

          小项目的完整文件的gittee链接

          • Tcp服务器架构:

            线程池模块

            #pragma once#include #include #include "log.hpp"#include #include #include templateclass RingQueue{private:    pthread_mutex_t Clock_;    pthread_mutex_t Plock_;    sem_t Psem_;    sem_t Csem_;    std::vector Queue_;    int Pptr_;    int Cptr_;    int capacity_;public:    RingQueue(int capacity = 10) : Queue_(capacity),Pptr_(0),Cptr_(0),capacity_(capacity){        sem_init(&Psem_,0,capacity);        sem_init(&Csem_,0,0);        pthread_mutex_init(&Clock_,nullptr);        pthread_mutex_init(&Plock_,nullptr);    }    ~RingQueue(){        sem_destroy(&Psem_);        sem_destroy(&Csem_);        pthread_mutex_destroy(&Clock_);        pthread_mutex_destroy(&Plock_);    }    T Pop(){        sem_wait(&Csem_);        pthread_mutex_lock(&Clock_);        T tem = Queue_[Cptr_];        Cptr_++;        Cptr_ %= capacity_;        pthread_mutex_unlock(&Clock_);        sem_post(&Psem_);        return tem;    }    void Push(T t){        sem_wait(&Psem_);        pthread_mutex_lock(&Plock_);        Queue_[Pptr_] = t;        Pptr_++;        Pptr_%= capacity_;        pthread_mutex_unlock(&Plock_);        sem_post(&Csem_);    }};

            #pragma once#include "sem_cp.cpp"#include #include #include #include #include "CalTask.cpp"templateclass Thread_Pool{    struct Thread_Data{        int Thread_num;        pthread_t tid;    };private:    RingQueue Queue_;  //线程安全的环形队列    std::vector thread_arr; //管理线程的容器    static std::mutex lock_;            //单例锁    static Thread_Pool * ptr_;    //单例指针private:    Thread_Pool(int capacity_Of_queue = 20) : Queue_(capacity_Of_queue){}    Thread_Pool(const Thread_Pool& Tp) = delete;    Thread_Pool& operator=(const Thread_Pool & Tp) = delete;public:    ~Thread_Pool(){}    //获取线程池单例-->注意C++的类模板静态成员函数需要在类体外进行定义    static Thread_Pool * Getinstance();    //创建多线程    void Create_thread(int thread_num = 10){        Thread_Data T_data;        for(int i = 0 ; i 该函数没有在类外调用,所以无须在类体外定义    static void* Routine(void * args){        Thread_Pool * Pool = static_cast(args);        while(true){            std::cout 用于执行具体任务            Thread_Task();        }        return nullptr;    }};//初始化静态指针templateThread_Pool * Thread_Pool::ptr_ = nullptr;templatestd::mutex Thread_Pool::lock_;//注意C++的类模板静态成员函数需要在类体外进行定义templateThread_Pool * Thread_Pool::Getinstance(){    if(ptr_ == nullptr){        lock_.lock();        if(ptr_ == nullptr){            ptr_ = new Thread_Pool;        }        lock_.unlock();    }    return ptr_;}

            序列化反序列化工具模块

            • 序列反序列化是保证通信过程中数据完整性的关键步骤,保证数据语义完整,结构完整

              #pragma once#include #include // 自定义序列化反序列化协议const std::string blank_space_sep = " ";const std::string protocol_sep = "\n";//封装报文std::string Encode(std::string &content){    //报文正文字节数    std::string package = std::to_string(content.size());    package += protocol_sep;    package += content;    //用分隔符封装正文    package += protocol_sep;    return package;}//解析报文package-->"正文长度"\n"正文"\nbool Decode(std::string &package, std::string& content){    size_t pos = package.find(protocol_sep);    if(pos == std::string::npos) return false;    //解析报文正文长度    size_t Len = std::atoi(package.substr(0,pos).c_str());    //确定报文是否完整    size_t total_Len = pos + Len + 2;    if(package.size() != total_Len) return false;    //获取正文内容    content = package.substr(pos+1,Len);    package.erase(0,total_Len);    return true;}//用户层协议请求结构体class Request{public:    int x;    int y;    char op; public:    Request(int data1 , int data2 , char op)        : x(data1),y(data2),op(op){}    Request(){}public:    //请求结构体 序列化 成报文正文字符串 "x op y"    bool Serialize(std::string& out){        std::string content = std::to_string(x);        content += blank_space_sep;        content += op;        content += blank_space_sep;        content += std::to_string(y);        out = content;        return true;        // 等价的jason代码        // Json::Value root;        // root["x"] = x;        // root["y"] = y;        // root["op"] = op;        // // Json::FastWriter w;        // Json::StyledWriter w;        // out = w.write(root);        // return true;    }    //报文正文字符串 反序列化 成请求结构体    // "x op y"    bool Deserialize(const std::string &in) {        size_t left = in.find(blank_space_sep);        if(left == std::string::npos)return false;        x = std::stoi(in.substr(0,left).c_str());        std::size_t right = in.rfind(blank_space_sep);        if (right == std::string::npos)return false;        y = std::atoi(in.substr(right + 1).c_str());        if(left + 2 != right) return false;        op = in[left+1];        return true;        // 等价的jason代码        // Json::Value root;        // Json::Reader r;        // r.parse(in, root);        // x = root["x"].asInt();        // y = root["y"].asInt();        // op = root["op"].asInt();        // return true;    }    void DebugPrint()    {        std::cout public:    int result;    int code; public:    Response(int res , int c)        : result(res),code(c){}    Response(){}public:    //请求回应结构体 序列化 成报文正文字符串 "result code"    bool Serialize(std::string& out){        std::string s = std::to_string(result);        s += blank_space_sep;        s += std::to_string(code);        out = s;        return true;        // 等价的jason代码        // Json::Value root;        // root["result"] = result;        // root["code"] = code;        // // Json::FastWriter w;        // Json::StyledWriter w;        // out = w.write(root);        // return true;    }    //"result code"    //报文正文字符串 反序列化 成请求回应结构体    bool Deserialize(const std::string &in)     {        std::size_t pos = in.find(blank_space_sep);        if (pos == std::string::npos)return false;        if(pos == 0 || pos == in.size() - 1) return false;        result = std::stoi(in.substr(0, pos).c_str());        code = std::stoi(in.substr(pos+1).c_str());        return true;        // 等价的jason代码        // Json::Value root;        // Json::Reader r;        // r.parse(in, root);        // result = root["result"].asInt();        // code = root["code"].asInt();        // return true;    }    void DebugPrint()    {        std::cout     //Tcp通讯构建器    class TcpServer{        enum{            UsageError = 1,            SocketError,            BindError,            ListenError,        };    private:        int socketfd_ = -1;        std :: string ip_;        uint16_t port_;        int backlog_ = 10;    public:        TcpServer(const std::string& ip = "172.19.29.44", uint16_t port = 8081) : ip_(ip) , port_(port){}        ~TcpServer(){if(socketfd_  0) close(socketfd_);}    public:        //确定通信协议,建立文件描述符        void BuildSocket(){            socketfd_ = socket(AF_INET,SOCK_STREAM,0);            if(socketfd_