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

文章目录

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

一.网络层与传输层协议

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

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

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

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

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

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

• Tcp服务器架构:
  

线程池模块

#pragma once
#include <iostream>
#include <pthread.h>
#include "log.hpp"
#include <semaphore.h>
#include <vector>
#include <cstdio>

template<class T>
class RingQueue{
private:
    pthread_mutex_t Clock_;
    pthread_mutex_t Plock_;
    sem_t Psem_;
    sem_t Csem_;
    std::vector<T> 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 <pthread.h>
#include <iostream>
#include <string>
#include <mutex>
#include "CalTask.cpp"

template<class Task>
class Thread_Pool{
    struct Thread_Data{
        int Thread_num;
        pthread_t tid;
    };
private:
    RingQueue<Task> Queue_;  //线程安全的环形队列
    std::vector<Thread_Data> thread_arr; //管理线程的容器
    static std::mutex lock_;            //单例锁
    static Thread_Pool<Task> * ptr_;    //单例指针
private:
    Thread_Pool(int capacity_Of_queue = 20) : Queue_(capacity_Of_queue){}
    Thread_Pool(const Thread_Pool<Task>& Tp) = delete;
    Thread_Pool<Task>& operator=(const Thread_Pool<Task> & Tp) = delete;
public:
    ~Thread_Pool(){}
    //获取线程池单例-->注意C++的类模板静态成员函数需要在类体外进行定义
    static Thread_Pool<Task> * Getinstance();
    //创建多线程
    void Create_thread(int thread_num = 10){
        Thread_Data T_data;
        for(int i = 0 ; i < thread_num ; ++i){
            //注意线程池对象的this指针传递给线程
            pthread_create(&T_data.tid,nullptr,Routine,this);
            T_data.Thread_num = i + 1;
            thread_arr.push_back(T_data);
        }
    }
    //线程等待
    void Thread_join(){
        for(int i = 0 ;i < thread_arr.size() ; ++i){
            pthread_join(thread_arr[i].tid,nullptr);
        }
    }
    //向线程池中加入任务
    void Push(Task T){
        Queue_.Push(T);
    }
    void Push(Task && T){
        Queue_.Push(std::forward<Task>(T));
    }
private:
    //线程函数-->该函数没有在类外调用,所以无须在类体外定义
    static void* Routine(void * args){
        Thread_Pool<Task> * Pool = static_cast<Thread_Pool<Task> *>(args);
        while(true){
            std::cout << "Thread prepare to work\n" << std::endl;
            Task Thread_Task = Pool->Queue_.Pop();
            //要求Task类重载()-->用于执行具体任务
            Thread_Task();
        }
        return nullptr;
    }
};


//初始化静态指针
template<class Task>
Thread_Pool<Task> * Thread_Pool<Task>::ptr_ = nullptr;
template<class Task>
std::mutex Thread_Pool<Task>::lock_;

//注意C++的类模板静态成员函数需要在类体外进行定义
template<class Task>
Thread_Pool<Task> * Thread_Pool<Task>::Getinstance(){
    if(ptr_ == nullptr){
        lock_.lock();
        if(ptr_ == nullptr){
            ptr_ = new Thread_Pool<Task>;
        }
        lock_.unlock();
    }
    return ptr_;
}

序列化反序列化工具模块

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

#pragma once
#include <iostream>
#include <string>

// 自定义序列化反序列化协议
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"正文"\n
bool 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 << "新请求构建完成:  " << x << op << y << "=?" << std::endl;
    }
};



//用户层协议请求回应结构体
class Response{
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 << "结果响应完成, result: " << result << ", code: "<< code << std::endl;
    }
};

通信信道建立模块

#pragma once
#include <iostream>
#include <string>
#include <sys/types.h>   
#include <sys/socket.h>
#include "log.hpp"
#include <memory.h>
#include <arpa/inet.h>
#include <netinet/in.h>


namespace MySocket{

    //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_ < 0){
                lg(Fatal,"socket error,%s\n",strerror(errno));
                exit(SocketError);
            }
        }
        //文件描述符与服务器ip : 端口号绑定
        void SocketBind(){
            struct sockaddr_in addr;
            memset(&addr,0,sizeof(addr));
            addr.sin_port = htons(port_);
            addr.sin_family = AF_INET;
            addr.sin_addr.s_addr = inet_addr(ip_.c_str());
            if(bind(socketfd_,(const sockaddr*)&addr,sizeof(addr)) < 0){
                lg(Fatal,"socket bind error,%s\n",strerror(errno));
                exit(BindError);
            }
            lg(Info,"socket bind success\n");
        }

        //启动服务监听,等待客户端的连接
        void Socklisten(){
            if(socketfd_ <= 0){
                lg(Fatal,"socket error,%s\n",strerror(errno));
                exit(SocketError);
            }
            if(listen(socketfd_,backlog_) < 0){
                lg(Fatal, "listen error, %s: %d", strerror(errno), errno);
                exit(ListenError);
            }
        }
        //服务器接收客户端的连接-->并创建用于通信的文件描述符-->一个客户端连接对应一个文件描述符
        int SockAccept(std::string& cilent_ip, uint16_t& cilent_port){
            struct sockaddr_in client_addr;  // 输出型参数,用于获取用户的ip : 端口号
            memset(&client_addr,0,sizeof(client_addr));
            socklen_t Len = sizeof(client_addr);
            int newfd = accept(socketfd_,(struct sockaddr*)&client_addr,&Len);
            if(newfd < 0){
                lg(Warning, "accept error, %s: %d", strerror(errno), errno);
                return -1;
            }
            //提取客户端信息-->输出参数
            char ipstr[64];
            cilent_ip = inet_ntop(AF_INET,&client_addr.sin_addr,ipstr,sizeof(ipstr));
            cilent_ip = ipstr;
            cilent_port = ntohs(client_addr.sin_port);
            return newfd;
        }


    public:
        int Get_Server_fd(){
            return socketfd_;
        }
        void Close_fd(){
            if(socketfd_ > 0){
                close(socketfd_);
                socketfd_ = -1;
            }
        }
    };
};

服务器主体模块

#pragma once
#include "ThreadPool.cpp"
#include "TcpServer.cpp"
#include "CalTask.cpp"
#include "log.hpp"
#include <signal.h>

//构建计算器服务器
class CalServer{
    const int size = 2048;
private:
    Thread_Pool<CalTask> * Pool_ptr_;
    MySocket::TcpServer Socket_;
    int Socket_fd_ = -1;
public:
    CalServer(const std::string& de_ip = "172.19.29.44",uint16_t de_port = 8081)
        : Socket_(de_ip,de_port)
    {
        Pool_ptr_ = Thread_Pool<CalTask>::Getinstance();
        if(Pool_ptr_ == nullptr){
            lg(Fatal,"Pool_ptr_ is nullptr\n");
            return;
        }
        Pool_ptr_->Create_thread();
    }
    ~CalServer(){}
public:
    //建立Tcp连接条件
    bool Init(){
        Socket_.BuildSocket();
        Socket_fd_ = Socket_.Get_Server_fd();
        if(Socket_fd_ < 0){
            lg(Fatal,"BuildSocket failed\n");
            return true;
        }
        Socket_.SocketBind();
        Socket_.Socklisten();
        lg(Info, "init server .... done");
        return true;
    }
    //启动服务器
    void Start(){
        signal(SIGCHLD, SIG_IGN);
        signal(SIGPIPE, SIG_IGN);
        char ReadBuffer[size];
        while(true){
            //接受用户请求
            std::string client_ip;
            uint16_t client_port;
            int client_fd = Socket_.SockAccept(client_ip,client_port);
            if(client_fd < 0){
                lg(Warning,"SockAccept error\n");
                continue;
            }
            lg(Info, "accept a new link, sockfd: %d, clientip: %s, clientport: %d", client_fd, client_ip.c_str(), client_port);

            int n = read(client_fd,ReadBuffer,sizeof(ReadBuffer));
            ReadBuffer[n] = 0;  
            std::string TaskStr(ReadBuffer);
            printf("receives mess from client : %s",ReadBuffer);
            if(n < 0){
                lg(Warning,"read error\n");
                break;
            }
            CalTask task(client_fd,client_ip,client_port,TaskStr);
            Pool_ptr_->Push(task);
        }
    }
};

任务回调模块(根据具体应用场景可重构)

#pragma once
#include <string>
#include "ThreadPool.cpp"
#include "Protocol.cpp"


enum{
    Div_Zero = 1,
    Mod_Zero,
    Other_Oper
};

class CalTask{
private:
    int socketfd_;                //网络通信文件描述符
    std :: string ip_;            //客户端ip
    uint16_t port_;               //客户端端口号
    std::string package_;         //客户请求字符串
public:
    CalTask(int socketfd,const std::string& ip , uint16_t & port,std::string & str)
        : socketfd_(socketfd),ip_(ip),port_(port),package_(str){}
    CalTask(){}//类一定要有默认构造函数
    ~CalTask(){}
public:

    //执行计算任务并将结果发送给用户
    void operator() (){
        std::cout << "Task Running ... \n" << std::endl;

        std::string content;
        //将用户发送的报文进行解包获取正文
        bool r = Decode(package_, content);
        if (!r)return;

        //将报文正文进行反序列化
        Request req;
        r = req.Deserialize(content);
        if (!r)return ;
        req.DebugPrint();

        content = ""; 
        //构建计算结果                         
        Response resp = CalculatorHelper(req);
        resp.DebugPrint();

        //计算结果序列化成字符串
        resp.Serialize(content);
        //字符串正文封装成报文发送给用户
        std::string ResStr = Encode(content);
        write(socketfd_,ResStr.c_str(),ResStr.size());

        if(socketfd_ > 0)close(socketfd_);
    }

private:
    Response CalculatorHelper(const Request &req){
        //构建请求回应结构体
        Response resp(0, 0);
        switch (req.op){
        case '+':
            resp.result = req.x + req.y;
            break;
        case '-':
            resp.result = req.x - req.y;
            break;
        case '*':
            resp.result = req.x * req.y;
            break;
        case '/':{
            if (req.y == 0)
                resp.code = Div_Zero;
            else
                resp.result = req.x / req.y;
        }
        break;
        case '%':{
            if (req.y == 0)
                resp.code = Mod_Zero;
            else
                resp.result = req.x % req.y;
        }
        break;
        default:
            resp.code = Other_Oper;
            break;
        }
        return resp;
    }
};

Tips:DebugC++代码过程中遇到的问题记录

• 使用C++类模板时,若在类模板中定义了静态成员函数,且该静态成员函数在类外被调用,则该静态成员函数必须定义在类外,不然链接器无法找到函数体.
• 注意类模板静态成员的声明格式需要加关键字temlpate<>
• 声明类模板静态成员时无需特化模版类型参数
• 跨主机并发通信测试: