[Linux理论基础1]----手写和使用json完成[序列化和反序列化]

文章目录

前言
一、应用层
二、再谈"协议"
三、网络版计算器

手写版本
使用第三方库json实现

完整代码
总结

前言

理解应用层的作用,初始HTTP协议;
理解传输层的作用,深入理解TCP的各项特性和机制;
对整个TCP/IP协议有系统的理解;
对TCP/IP协议体系下的其他重要协议和技术有一定的了解;
学会使用一些分析网络问题的工具和方法;

本章属于Linux下网络编程的理论基础.

正文开始!

一、应用层

我们程序猿写的一个个解决我们实际问题,满足我们日常需求的网络程序,都是在应用层.

[Linux理论基础1]----手写和使用json完成[序列化和反序列化]

二、再谈"协议"

协议是一种"约定".socket api的接口在读写数据的时候,都是按"字符串"的方式来发送接收的.如果我们要传输一些"结构化的数据"怎么办呢?

三、网络版计算器

例如,我们需要实现一个服务器版的加法器,我们需要客户端把要计算的两个加数发过去,然后由服务器进行计算,最后再把结果返回给客户端.
约定方案一:

客户端发送一个形如"1+1"的字符串;
这个字符串中有两个操作数,都是整形;
两个数字之间会有一个字符是运算符,运算符只能是 + ;
数字和运算符之间没有空格;
…

[Linux理论基础1]----手写和使用json完成[序列化和反序列化]

约定方案二:

定义结构体来表示我们需要交互的信息;
发送数据是将这个结构体按照一个规则转换成为字符串,接收到数据的时候再按照相同的规则把字符串转化回结构体;
这个过程叫做"序列化"和"反序列化";

手写版本

makefile

.PHONY:all
all:clientTcp serverTcpd
clientTcp:clientTcp.cc
  g++ -o $@ $^ $(Method) -std=c++11 -ljsoncpp
serverTcpd:serverTcp.cc
  g++ -o $@ $^ $(Method) -std=c++11 -ljsoncpp -lpthread
.PHONY:clean
clean:
  rm -rf clientTcp serverTcpd ServerTcp.log

clientTcp.cc

#include "util.hpp"
#include"Protocol.hpp"
#include "log.hpp"

using namespace std;

volatile bool quit =false;

static void Usage(string proc)
{
    cerr << "Usage\n\t" << proc << " ip port" << endl;
    cerr << "Example\n\t" << proc << " 127.0.0.1 8080\n"
         << endl;
}

// 2.需要bind吗??--->需要，但是不需要显式的bind！
// 3.需要listen吗？不需要的！
// 4.需要accept吗？不需要的！
// ./clientTcp serverIp serverPort
int main(int argc, char *argv[])
{
    if (argc != 2 && argc != 3)
    {
        Usage(argv[0]);
        exit(USAGE_ERR);
    }
    string serverIp = argv[1];
    uint16_t serverPort = stoi(argv[2]);

    // 1.创建套接字
    int sock = socket(AF_INET, SOCK_STREAM, 0);
    if (sock < 0)
    {
        cerr << "socket: " << strerror(errno) << endl;
        exit(SOCKET_ERR);
    }

    // 2.connect,发起连接请求，你想谁发起请求呢？当然是想服务器发起请求喽
    // 2.1 先填充需要连接的远端主机的基本信息
    struct sockaddr_in server;
    memset(&server, 0, sizeof(server));
    server.sin_family = AF_INET;
    inet_aton(serverIp.c_str(), &server.sin_addr);
    server.sin_port = ntohs(serverPort);
    // 2.2发送请求，connect会自动帮我们进行bind！
    if (connect(sock, (const sockaddr *)&server, sizeof(server)) != 0)
    {
        cerr << "connect: " << strerror(errno) << endl;
        exit(CONN_ERR);
    }
    cout << "info connect success: " << sock << endl;
    //trimStr(message); //对多输入的空格进行清洗
    string message;
    while(!quit)
    {
        cout<<"请输入表达式>> "; // 1 + 1
        getline(cin,message);
        if(strcasecmp(message.c_str(),"quit")==0)
        {    
            quit=true;
            continue;
        }     
        Request req;
        if(!makeRequest(message,&req))
            continue;
        std::string package;
        req.serialize(&package);
        cout<<"debug->serialize-> \n"<<package<<endl;
        package = encode(package,package.size());
        cout<<"debug->encode-> \n"<<package<<endl;
        ssize_t s=write(sock,package.c_str(),package.size());
        if(s>0)
        {
            char buffer[BUFFER_SIZE];
            size_t s =read(sock,buffer,sizeof(buffer)-1);
            if(s > 0) buffer[s]=0;
            cout<<buffer<<endl;
            string echoPackage = buffer;
            Response resp;
            uint32_t len=0;
            cout<<"debug->get response-> "<<echoPackage<<endl;
            string tmp = decode(echoPackage,&len);
            if(len > 0)
            {
                echoPackage = tmp;
                cout<<"debug->decode-> "<<echoPackage<<endl;
                resp.deserialize(echoPackage);
                printf("[exitcode: %d] %d\n",resp._exitCode,resp._result);
            }
        }
        else if(s<=0)
        {
            break;
        }
        message.clear();
    }
    close(sock);
    return 0;
}

calServer.cc

#include "util.hpp"
#include "log.hpp"
#include "ThreadPool.hpp"
#include "Protocol.hpp"
#include "Task.hpp"
#include "daemonize.hpp"
using namespace std;

class ServerTcp;
struct ThreadData
{
    ThreadData(int sock, string clientIp, uint16_t clientPort, ServerTcp *ts)
        : _sock(sock), _clientIp(clientIp), _clientPort(clientPort), _this(ts)
    {
    }
    int _sock;
    string _clientIp;
    uint16_t _clientPort;
    ServerTcp *_this;
};
static Response calcaulator(const Request &req)
{
    Response resp;
    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._exitCode = 1; // 1:除0
            else
                resp._result = req._x / req._y;
        }
        break;
    case '%':
        {
            if(req._y == 0)
                resp._exitCode = 2; // 1:模0
            else
                resp._result = req._x % req._y;
        }
        break;
    default:
        resp._exitCode = 3; //非法操作符
        break;
    }
    return resp;
}

// 1. 全部手写
// 2.部分采用别人的方案 -- 序列化和反序列化的问题  -- xml,json,protobuf 
void netCal(int sock, const string &clientIp, uint16_t clientPort)
{
    assert(sock > 0);
    assert(!clientIp.empty());
    assert(clientPort > 1024);

    // 9\r\n100 + 200\r\n
    string inbuffer;
    while (true)
    {
        char buff[128];
        ssize_t s = read(sock, buff, sizeof(buff) - 1);
        if (s == 0)
        {
            logMessage(NOTICE, "client[%s:%d] close sock, service done", clientIp.c_str(), clientPort);
            break;
        }
        else if (s < 0)
        {
            logMessage(WARINING, "read client[%s:%d] error, errcode: %d, errormessage: %s",
                       clientIp.c_str(), clientPort, errno, strerror(errno));
            break;
        }
        cout<<buff<<endl;
        // read success
        buff[s] = 0;
        inbuffer += buff;
        // 1.检查inbuffer是不是已经具有了一个strPackage
        Request req;
        uint32_t packageLen = 0;
        string package = decode(inbuffer, &packageLen);
        if (packageLen == 0)
            continue; // 无法提取一个完整的报文,进行下一次读取
        // 2.证明已经获得一个完整的package
        if (req.deserialize(package))
        {
            req.debug();
            // 3.处理逻辑,输入一个req,得到一个resp
            Response resp = calcaulator(req); //resp是一个结构化的字段
            // 4.对resp进行序列化
            string respPackage;
            resp.serialize(&respPackage);
            // 5.对报文进行encode
            respPackage = encode(respPackage,respPackage.size());
            // 6.简单进行发送
            write(sock,respPackage.c_str(),respPackage.size());

        }
    }
}

class ServerTcp
{
public:
    ServerTcp(uint16_t port, string ip = "")
        : _listenSock(-1), _port(port), _ip(ip), _tp(nullptr)
    {
        _quit = false;
    }
    ~ServerTcp()
    {
    }

public:
    void init()
    {
        // 1.创建socket
        _listenSock = socket(AF_INET, SOCK_STREAM, 0);
        if (_listenSock < 0)
        {
            logMessage(FATAL, "socket:%s", strerror(errno));
            exit(SOCKET_ERR);
        }
        logMessage(DEBUG, "socket:%s,%d", strerror(errno), _listenSock);
        // 2.bind绑定
        // 2.1填充服务器
        struct sockaddr_in local; // 用户栈
        memset(&local, 0, sizeof local);
        local.sin_family = AF_INET;
        local.sin_port = htons(_port);
        _ip.empty() ? (local.sin_addr.s_addr = INADDR_ANY) : (inet_aton(_ip.c_str(), &local.sin_addr));
        // 2.2本地socket信息，写入_sock对应的内核区域
        if (bind(_listenSock, (const sockaddr *)&local, sizeof local) < 0)
        {
            logMessage(FATAL, "bind: %s", strerror(errno));
            exit(BIND_ERR);
        }
        logMessage(DEBUG, "bind: %s", strerror(errno));
        // 3.监听socket，为何要监听呢？tcp是面向连接的！
        if (listen(_listenSock, 5 /*后面再说*/) < 0)
        {
            logMessage(FATAL, "listen: %s", strerror(errno));
            exit(LISTEN_ERR);
        }
        logMessage(DEBUG, "listen: %s", strerror(errno));
        // 允许别人来连接你了

        // 4.加载线程池
        _tp = ThreadPool<Task>::getInstance();
    }
    void loop()
    {
        // signal(SIGCHLD,SIG_IGN);//只在Linux下有效
        _tp->start();
        logMessage(DEBUG, "thread pool start success,thread num: %d", _tp->ThreadNum());

        while (!_quit)
        {
            struct sockaddr_in peer;
            socklen_t len = sizeof(peer);
            // 4.获取连接,accept的返回值是一个新的socket fd？？
            // 4.1 _listenScok:监听&&获取新的连接--->sock
            // 4.2 serviceSock:给用户提供新的socket服务
            int serviceSock = accept(_listenSock, (struct sockaddr *)&peer, &len);
            if (_quit)
                break;
            if (serviceSock < 0)
            {
                // 获取连接失败
                logMessage(WARINING, "accept: &s[%d]", strerror(errno), serviceSock);
                continue;
            }
            // 4.1获取客户端基本信息
            uint16_t peerPort = ntohs(peer.sin_port);
            string peerIp = inet_ntoa(peer.sin_addr);
            logMessage(DEBUG, "accept: %s | %s[%d],socker fd: %d",
                       strerror(errno), peerIp.c_str(), peerPort, serviceSock);

            // 5.提供服务，小写转大写
            Task t(serviceSock, peerIp, peerPort, netCal);
            _tp->push(t);

            // logMessage(DEBUG,"server provide service start ...");
            // sleep(1);
        }
    }

    bool quitServer()
    {
        _quit = true;
    }

private:
    int _listenSock;
    uint16_t _port;
    string _ip;
    // 引入线程池
    ThreadPool<Task> *_tp;
    // 安全退出
    bool _quit;
};
static void Usage(string proc)
{
    cerr << "Usage\n\t" << proc << " port ip" << endl;
    cerr << "Example\n\t" << proc << " 8080  127.0.0.1\n"
         << endl;
}

ServerTcp *svrp = nullptr;

void sighandler(int signo)
{
    if (signo == 3 && svrp != nullptr)
    {
        svrp->quitServer();
    }
    logMessage(DEBUG, "server quit save!");
}

// ./serverTcp local_port [local_ip]
int main(int argc, char *argv[])
{
    if (argc != 2 && argc != 3)
    {
        Usage(argv[0]);
        exit(USAGE_ERR);
    }
    uint16_t port = stoi(argv[1]);
    string ip;
    if (argc == 3)
    {
        ip = argv[2];
    }
    signal(3, sighandler);
    ServerTcp svr(port, ip);
    svr.init();
    svrp = &svr;
    svr.loop();
    return 0;
}

protocol.hpp

#pragma once
#include <iostream>
#include <string>
#include <cstdio>
#include <cassert>
#include <cstring>
#include "util.hpp"
#include<jsoncpp/json/json.h>
using namespace std;
// 我们要在这里进行我们自己的协议定制！
// 网络版本的计算机

#define CRLF "\r\n"
#define CRLF_LEN strlen(CRLF)
#define SPACE " "
#define SPACE_LEN strlen(SPACE)
#define OPS "+-*/%"

// encode,整个序列化之后的字符串进行添加长度
// strlen(四个字节)XXXXXXXX
// "strlen\r\n"XXXXXXXX\r\n --采用这种方案
// encode,整个序列化之后的字符串进行添加长度
std::string encode(const std::string &in, uint32_t len)
{
    string encodein = to_string(len);
    encodein += CRLF;
    encodein += in;
    encodein += CRLF;
    return encodein;
}

// decode，整个序列化之后的字符串进行提取长度
// 1.必须具有完整的长度
// 2.必须具有和len相符合的有效载荷
// 我们才返回有效载荷和len
// 否则我们就是一个检测函数!
// 9\r\n100 + 200\r\n 9\r\n100 + 200\r\n
std::string decode(std::string &in, uint32_t *len)
{
    assert(len);
    // 1.确认是否是一个包含len的有效字符串
    *len = 0;
    size_t pos = in.find(CRLF);
    if (pos == string::npos)
        return "";
    // 2.提取长度
    string inLen = in.substr(0, pos);
    int intLen = stoi(inLen);
    // 3.确认有效载荷也是符合要求的
    int surplus = in.size() - 2 * CRLF_LEN - pos;
    if (surplus < intLen)
        return "";
    // 4.确认有完整的报文结构
    std::string package = in.substr(pos + CRLF_LEN, intLen);
    *len = intLen;
    // 5.将当前的报文完整的从in中全部移除掉!
    long removeLen = 2 * CRLF_LEN + inLen.size() + package.size();
    in.erase(0, removeLen);
    // 6.正常返回
    return package;
}

// 定制的请求 x op y
class Request
{
public:
    Request()
    {
    }
    ~Request()
    {
    }
    // 序列化 -- 结构化的数据 -> 字符串
    // 认为结构化字段中的内容已经被填充了
    void serialize(std::string *out)
    {
        *out = to_string(_x) + SPACE + _op + SPACE + to_string(_y);
    }

    // 反序列化 -- 字符串 -> 结构化的数据
    //  9\r\n100 + 200\r\n -> 100 + 200
    bool deserialize(std::string &in)
    {
        size_t spaceOne = in.find(SPACE);
        if (spaceOne == string::npos)
            return false;
        size_t spaceTwo = in.rfind(SPACE);
        if (spaceTwo == string::npos)
            return false;

        string dataOne = in.substr(0, spaceOne);
        string dataTwo = in.substr(spaceTwo + SPACE_LEN);
        string oper = in.substr(spaceOne + SPACE_LEN, spaceTwo - spaceOne - SPACE_LEN);
        if (oper.size() != 1)
            return false;

        // 转成内部成员
        _x = stoi(dataOne);
        _y = stoi(dataTwo);
        _op = oper[0];
        return true;
    }
    void debug()
    {
        cout << "#####################################" << endl;
        printf("_x=%d,_op=%c,_y=%d\n", _x, _op, _y);
        cout << "#####################################" << endl;
    }

public:
    // 需要计算的数据
    int _x;
    int _y;
    // 需要计算的种类
    char _op; //+ - * / %
};

// 定制的响应2
class Response
{
public:
    Response() : _exitCode(0), _result(0)
    {
    }
    ~Response()
    {
    }
    // 序列化 -- 不仅仅是在网络中应用,本地也是可以直接使用的
    //  "_exitCode _result"
    void serialize(std::string *out)
    {
        string ec = to_string(_exitCode);
        string res = to_string(_result);
        *out = ec + SPACE + res;
    }
    // 反序列化
    bool deserialize(const std::string &in)
    {
        size_t pos = in.find(SPACE);
        if (pos == string::npos)
            return false;
        string codeStr = in.substr(0, pos);
        string resStr = in.substr(pos + SPACE_LEN);
        // 将反序列化的结果写入到内部成员中,形成结构化数据.
        _exitCode = stoi(codeStr);
        _result = stoi(resStr);
        return true;
    }
    void debug()
    {
        cout << "#####################################" << endl;
        printf("_exitCode=%d,_result=%d\n", _exitCode, _result);
        cout << "#####################################" << endl;
    }

public:
    // 退出状态，0标识运算结果合法，非0标识运算结果是非法的，！0是几就表示是什么原因错了！
    int _exitCode;

    int _result;
};

bool makeRequest(string &str, Request *req)
{
    char strtmp[BUFFER_SIZE];
    snprintf(strtmp, sizeof strtmp, "%s", str.c_str());
    char *left = strtok(strtmp, OPS);
    if (!left)
        return false;
    char *right = strtok(nullptr, OPS);
    if (!right)
        return false;

    char mid = str[strlen(left)];
    req->_x = stoi(left);
    req->_y = stoi(right);
    req->_op = mid;
    return true;
}

[Linux理论基础1]----手写和使用json完成[序列化和反序列化]

使用第三方库json实现

相比于手写版本的代码,使用库的版本只进行修改了makefile和protocol.hpp两个文件!

云服务器中安装json库

sudo yum install -y jsoncpp-devel

和线程库一样,json也属于第三方库,所以在makefile编译中需要带-ljsoncpp选项,否则编译就会失败!

makefile

.PHONY:all
all:clientTcp serverTcpd
Method=#-DMY_SELF #进行命令行宏定义,如果定义则编译手写版本,否则编译第三方库版本
clientTcp:clientTcp.cc
  g++ -o $@ $^ $(Method) -std=c++11 -ljsoncpp
serverTcpd:serverTcp.cc
  g++ -o $@ $^ $(Method) -std=c++11 -ljsoncpp -lpthread
.PHONY:clean
clean:
  rm -rf clientTcp serverTcpd ServerTcp.log

protocol.hpp

#pragma once
#include <iostream>
#include <string>
#include <cstdio>
#include <cassert>
#include <cstring>
#include "util.hpp"
#include<jsoncpp/json/json.h>
using namespace std;
// 我们要在这里进行我们自己的协议定制！
// 网络版本的计算机

#define CRLF "\r\n"
#define CRLF_LEN strlen(CRLF)
#define SPACE " "
#define SPACE_LEN strlen(SPACE)
#define OPS "+-*/%"

// encode,整个序列化之后的字符串进行添加长度
// strlen(四个字节)XXXXXXXX
// "strlen\r\n"XXXXXXXX\r\n --采用这种方案
// encode,整个序列化之后的字符串进行添加长度
std::string encode(const std::string &in, uint32_t len)
{
    string encodein = to_string(len);
    encodein += CRLF;
    encodein += in;
    encodein += CRLF;
    return encodein;
}

// decode，整个序列化之后的字符串进行提取长度
// 1.必须具有完整的长度
// 2.必须具有和len相符合的有效载荷
// 我们才返回有效载荷和len
// 否则我们就是一个检测函数!
// 9\r\n100 + 200\r\n 9\r\n100 + 200\r\n
std::string decode(std::string &in, uint32_t *len)
{
    assert(len);
    // 1.确认是否是一个包含len的有效字符串
    *len = 0;
    size_t pos = in.find(CRLF);
    if (pos == string::npos)
        return "";
    // 2.提取长度
    string inLen = in.substr(0, pos);
    int intLen = stoi(inLen);
    // 3.确认有效载荷也是符合要求的
    int surplus = in.size() - 2 * CRLF_LEN - pos;
    if (surplus < intLen)
        return "";
    // 4.确认有完整的报文结构
    std::string package = in.substr(pos + CRLF_LEN, intLen);
    *len = intLen;
    // 5.将当前的报文完整的从in中全部移除掉!
    long removeLen = 2 * CRLF_LEN + inLen.size() + package.size();
    in.erase(0, removeLen);
    // 6.正常返回
    return package;
}

// 定制的请求 x op y
class Request
{
public:
    Request()
    {
    }
    ~Request()
    {
    }
    // 序列化 -- 结构化的数据 -> 字符串
    // 认为结构化字段中的内容已经被填充了
    void serialize(std::string *out)
    {
#ifdef MY_SELF
        *out = to_string(_x) + SPACE + _op + SPACE + to_string(_y);
#else
        //json
        // 1.Value对象,万能对象,
        // 2.json是基于KV
        // 3.json有两套操作方法
        // 4.序列化的时候,会将所有的数据内容,转化称为字符串
        Json::Value root;
        root["x"]=_x;
        root["y"]=_y;
        root["op"]=_op;

        // Json::FastWriter fw;
        Json::StyledWriter fw;
        *out = fw.write(root);
#endif
    }

    // 反序列化 -- 字符串 -> 结构化的数据
    //  9\r\n100 + 200\r\n -> 100 + 200
    bool deserialize(std::string &in)
    {
#ifdef MY_SELF
        size_t spaceOne = in.find(SPACE);
        if (spaceOne == string::npos)
            return false;
        size_t spaceTwo = in.rfind(SPACE);
        if (spaceTwo == string::npos)
            return false;

        string dataOne = in.substr(0, spaceOne);
        string dataTwo = in.substr(spaceTwo + SPACE_LEN);
        string oper = in.substr(spaceOne + SPACE_LEN, spaceTwo - spaceOne - SPACE_LEN);
        if (oper.size() != 1)
            return false;

        // 转成内部成员
        _x = stoi(dataOne);
        _y = stoi(dataTwo);
        _op = oper[0];
        return true;

#else
        //json
        Json::Value root;
        Json::Reader rd;
        rd.parse(in,root);
        _x = root["x"].asInt();
        _y = root["y"].asInt();
        _op = root["op"].asInt();
        return true;
#endif
    }
    void debug()
    {
        cout << "#####################################" << endl;
        printf("_x=%d,_op=%c,_y=%d\n", _x, _op, _y);
        cout << "#####################################" << endl;
    }

public:
    // 需要计算的数据
    int _x;
    int _y;
    // 需要计算的种类
    char _op; //+ - * / %
};

// 定制的响应2
class Response
{
public:
    Response() : _exitCode(0), _result(0)
    {
    }
    ~Response()
    {
    }
    // 序列化 -- 不仅仅是在网络中应用,本地也是可以直接使用的
    //  "_exitCode _result"
    void serialize(std::string *out)
    {
#ifdef MY_SELF
        string ec = to_string(_exitCode);
        string res = to_string(_result);

        *out = ec + SPACE + res;
#else
        //json
        Json::Value root;
        root["code"]=_exitCode;
        root["res"]=_result;
        //Json::FastWriter fw;
        Json::StyledWriter fw;
        *out = fw.write(root);
#endif
    }
    // 反序列化
    bool deserialize(const std::string &in)
    {
#ifdef MY_SELF
        size_t pos = in.find(SPACE);
        if (pos == string::npos)
            return false;
        string codeStr = in.substr(0, pos);
        string resStr = in.substr(pos + SPACE_LEN);
        // 将反序列化的结果写入到内部成员中,形成结构化数据.
        _exitCode = stoi(codeStr);
        _result = stoi(resStr);
        return true;
#else
        //json
        Json::Value root;
        Json::Reader rd;
        rd.parse(in,root);
        _exitCode = root["code"].asInt();
        _result = root["res"].asInt();
        return true;
#endif
    }
    void debug()
    {
        cout << "#####################################" << endl;
        printf("_exitCode=%d,_result=%d\n", _exitCode, _result);
        cout << "#####################################" << endl;
    }

public:
    // 退出状态，0标识运算结果合法，非0标识运算结果是非法的，！0是几就表示是什么原因错了！
    int _exitCode;

    int _result;
};

bool makeRequest(string &str, Request *req)
{
    char strtmp[BUFFER_SIZE];
    snprintf(strtmp, sizeof strtmp, "%s", str.c_str());
    char *left = strtok(strtmp, OPS);
    if (!left)
        return false;
    char *right = strtok(nullptr, OPS);
    if (!right)
        return false;

    char mid = str[strlen(left)];
    req->_x = stoi(left);
    req->_y = stoi(right);
    req->_op = mid;
    return true;
}

[Linux理论基础1]----手写和使用json完成[序列化和反序列化]

在这里我们没有进行删除手写版本的序列化和反序列化,通过命令行进行宏定义和条件编译来进行版本的切换.

完整代码

lock.hpp

#pragma once

#include <iostream>
#include <pthread.h>

class Mutex
{
public:
    Mutex()
    {
        pthread_mutex_init(&_lock, nullptr);
    }

    void lock()
    {
        pthread_mutex_lock(&_lock);
    }
    void unlock()
    {
        pthread_mutex_unlock(&_lock);
    }
    ~Mutex()
    {
        pthread_mutex_destroy(&_lock);
    }

private:
    pthread_mutex_t _lock;
};

class LockGuard
{
public:
    LockGuard(Mutex* mutex)
        : _mutex(mutex)
    {
        _mutex->lock();
        std::cout<<"加锁成功..."<<std::endl;
    }
    ~LockGuard()
    {
        _mutex->unlock();
        std::cout<<"解锁成功..."<<std::endl;
    }

private:
    Mutex* _mutex;
};

log.hpp

#pragma once
#include<cstdio>
#include<ctime>
#include<cstdarg>
#include<cassert>
#include<cstdlib>
#include<cstring>
#include<cerrno>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#define DEBUG     0
#define NOTICE    1
#define WARINING  2
#define FATAL     3

const char* log_level[]={"DEBUG","NOTICE","WARINING","FATAL"};

#define LOGFIFE "ServerTcp.log"

class Log
{
public:
    Log():_logFd((-1))
    {}
    ~Log()
    {
        if(_logFd!=-1)
        {
            fsync(_logFd);//将操作系统中的数据尽快刷盘
            close(_logFd);
        }

    }
    void enable()
    {
        _logFd=open(LOGFIFE,O_WRONLY|O_APPEND|O_CREAT,0666);
        assert(_logFd!=-1);
        dup2(_logFd,0);
        dup2(_logFd,1);
        dup2(_logFd,2);
    }
private:
    int _logFd;
};




//logMessage(DEBUG,"%d",10);
void logMessage(int level,const char* format,...)
{
    assert(level>=DEBUG);
    assert(level<=FATAL);
    char logInfo[1024];
    char* name=getenv("USER");
    va_list ap; //ap--->char*
    va_start(ap,format);

    vsnprintf(logInfo,sizeof(logInfo)-1,format,ap);

    va_end(ap); //ap=NULL

    FILE* out=(level==FATAL)?stderr:stdout;
    fprintf(out,"%s | %u | %s | %s\n",\
    log_level[level],(unsigned int)time(nullptr),\
    name==nullptr?"unknow":name,logInfo);
    fflush(out);//将C缓冲区中的数据刷新到OS

}

Task.hpp

#pragma once

#include <iostream>
#include <string>
#include<functional>
#include"log.hpp"
#include<pthread.h>
class Task
{
public:

    //等价于
    // typedef std::function<void(int,std::string,uint16_t)> callback_t;
    using callback_t=std::function<void (int, std::string, uint16_t)>;

private:
    int _sock;//给用户提供任务IO服务的sock
    std::string _ip;//client ip
    uint16_t _port;//client port
    callback_t _func;//回调方法
public:
    Task():_sock(-1),_port(-1)
    {}
    Task(int sock,std::string ip,uint16_t port,callback_t func)
    :_sock(sock),_ip(ip),_port(port),_func(func)
    {}
    void operator()()
    {
        logMessage(DEBUG,"线程ID[%p]->处理%s:%d的请求 开始啦....",\
            pthread_self(),_ip.c_str(),_port);

        _func(_sock,_ip,_port);

        logMessage(DEBUG,"线程ID[%p]->处理%s:%d的请求 结束啦....",\
            pthread_self(),_ip.c_str(),_port);
    }

    ~Task()
    {}
};

ThreadPool.hpp

#pragma once

#include <iostream>
#include <queue>
#include <cassert>
#include <cstdlib>
#include <unistd.h>
#include <pthread.h>
#include<sys/prctl.h>
#include "Task.hpp"
#include "lock.hpp"
using namespace std;

const int gThreadNum = 15;

//设计为懒汉模式
template <class T>
class ThreadPool
{
private:
    ThreadPool(int threadNum = gThreadNum)
        : _threadNum(threadNum), _isStart(false)
    {
        assert(_threadNum > 0);
        pthread_mutex_init(&_mutex, nullptr);
        pthread_cond_init(&_cond, nullptr);
    }
    ThreadPool(const ThreadPool<T> &) = delete;
    void operator=(const ThreadPool<T> &) = delete;

public:
    static ThreadPool<T> *getInstance()
    {
        static Mutex mutex;
        if (nullptr == instance)//仅仅是过滤重复的判断
        {
            LockGuard lockguard(&mutex);//进入代码块，加锁，退出代码块，自动解锁
            if (nullptr == instance)
            {
                instance = new ThreadPool<T>();
            }
        }
        return instance;
    }
    ~ThreadPool()
    {
        pthread_mutex_destroy(&_mutex);
        pthread_cond_destroy(&_cond);
    }
    int ThreadNum()
    {
        return _threadNum;
    }

    //类内成员，成员函数都有默认参数this
    static void *threadRoutine(void *args)
    {
        pthread_detach(pthread_self());
        ThreadPool<T> *tp = static_cast<ThreadPool<T> *>(args);
        while (true)
        {
            tp->lockQueue();
            while (!tp->haveTask())
            {
                tp->waitForTask();
            }
            T t = tp->pop();
            tp->unlockQueue();
            t();//让指定的线程处理这个任务

        }
    }
    void start()
    {
        assert(!_isStart);
        for (int i = 0; i < _threadNum; i++)
        {
            pthread_t tmp;
            pthread_create(&tmp, nullptr, threadRoutine, this);
        }
        _isStart = true;
    }
    void push(const T &in)
    {
        lockQueue();
        _taskQueue.push(in);
        choiceThreadForHandler();
        unlockQueue();
    }

private:
    void lockQueue() { pthread_mutex_lock(&_mutex); }
    void unlockQueue() { pthread_mutex_unlock(&_mutex); }
    bool haveTask() { return !_taskQueue.empty(); }
    void waitForTask() { pthread_cond_wait(&_cond, &_mutex); }
    void choiceThreadForHandler() { pthread_cond_signal(&_cond); }
    T pop()
    {
        T tmp = _taskQueue.front();
        _taskQueue.pop();
        return tmp;
    }

private:
    bool _isStart;
    int _threadNum;
    queue<T> _taskQueue;
    pthread_mutex_t _mutex;
    pthread_cond_t _cond;
    static ThreadPool<T> *instance;
};

template <class T>
ThreadPool<T> *ThreadPool<T>::instance = nullptr;

util.hpp

#include <iostream>
#include <string>
#include<cstring>
#include<stdio.h>
#include<cstdlib>
#include <ctype.h>

#include<signal.h>
#include<unistd.h>
#include<sys/wait.h>
#include<sys/types.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<arpa/inet.h> 
#include<pthread.h>

#define SOCKET_ERR   1
#define BIND_ERR     2
#define LISTEN_ERR   3
#define USAGE_ERR    4
#define CONN_ERR     5

#define BUFFER_SIZE 1024

makefile

.PHONY:all
all:clientTcp serverTcpd
Method=#-DMY_SELF
clientTcp:clientTcp.cc
  g++ -o $@ $^ $(Method) -std=c++11 -ljsoncpp
serverTcpd:serverTcp.cc
  g++ -o $@ $^ $(Method) -std=c++11 -ljsoncpp -lpthread
.PHONY:clean
clean:
  rm -rf clientTcp serverTcpd ServerTcp.log

protocol.hpp

#pragma once
#include <iostream>
#include <string>
#include <cstdio>
#include <cassert>
#include <cstring>
#include "util.hpp"
#include<jsoncpp/json/json.h>
using namespace std;
// 我们要在这里进行我们自己的协议定制！
// 网络版本的计算机

#define CRLF "\r\n"
#define CRLF_LEN strlen(CRLF)
#define SPACE " "
#define SPACE_LEN strlen(SPACE)
#define OPS "+-*/%"

// encode,整个序列化之后的字符串进行添加长度
// strlen(四个字节)XXXXXXXX
// "strlen\r\n"XXXXXXXX\r\n --采用这种方案
// encode,整个序列化之后的字符串进行添加长度
std::string encode(const std::string &in, uint32_t len)
{
    string encodein = to_string(len);
    encodein += CRLF;
    encodein += in;
    encodein += CRLF;
    return encodein;
}

// decode，整个序列化之后的字符串进行提取长度
// 1.必须具有完整的长度
// 2.必须具有和len相符合的有效载荷
// 我们才返回有效载荷和len
// 否则我们就是一个检测函数!
// 9\r\n100 + 200\r\n 9\r\n100 + 200\r\n
std::string decode(std::string &in, uint32_t *len)
{
    assert(len);
    // 1.确认是否是一个包含len的有效字符串
    *len = 0;
    size_t pos = in.find(CRLF);
    if (pos == string::npos)
        return "";
    // 2.提取长度
    string inLen = in.substr(0, pos);
    int intLen = stoi(inLen);
    // 3.确认有效载荷也是符合要求的
    int surplus = in.size() - 2 * CRLF_LEN - pos;
    if (surplus < intLen)
        return "";
    // 4.确认有完整的报文结构
    std::string package = in.substr(pos + CRLF_LEN, intLen);
    *len = intLen;
    // 5.将当前的报文完整的从in中全部移除掉!
    long removeLen = 2 * CRLF_LEN + inLen.size() + package.size();
    in.erase(0, removeLen);
    // 6.正常返回
    return package;
}

// 定制的请求 x op y
class Request
{
public:
    Request()
    {
    }
    ~Request()
    {
    }
    // 序列化 -- 结构化的数据 -> 字符串
    // 认为结构化字段中的内容已经被填充了
    void serialize(std::string *out)
    {
#ifdef MY_SELF
        *out = to_string(_x) + SPACE + _op + SPACE + to_string(_y);
#else
        //json
        // 1.Value对象,万能对象,
        // 2.json是基于KV
        // 3.json有两套操作方法
        // 4.序列化的时候,会将所有的数据内容,转化称为字符串
        Json::Value root;
        root["x"]=_x;
        root["y"]=_y;
        root["op"]=_op;

        Json::FastWriter fw;
        // Json::StyledWriter fw;
        *out = fw.write(root);
#endif
    }

    // 反序列化 -- 字符串 -> 结构化的数据
    //  9\r\n100 + 200\r\n -> 100 + 200
    bool deserialize(std::string &in)
    {
#ifdef MY_SELF
        size_t spaceOne = in.find(SPACE);
        if (spaceOne == string::npos)
            return false;
        size_t spaceTwo = in.rfind(SPACE);
        if (spaceTwo == string::npos)
            return false;

        string dataOne = in.substr(0, spaceOne);
        string dataTwo = in.substr(spaceTwo + SPACE_LEN);
        string oper = in.substr(spaceOne + SPACE_LEN, spaceTwo - spaceOne - SPACE_LEN);
        if (oper.size() != 1)
            return false;

        // 转成内部成员
        _x = stoi(dataOne);
        _y = stoi(dataTwo);
        _op = oper[0];
        return true;

#else
        //json
        Json::Value root;
        Json::Reader rd;
        rd.parse(in,root);
        _x = root["x"].asInt();
        _y = root["y"].asInt();
        _op = root["op"].asInt();
        return true;
#endif
    }
    void debug()
    {
        cout << "#####################################" << endl;
        printf("_x=%d,_op=%c,_y=%d\n", _x, _op, _y);
        cout << "#####################################" << endl;
    }

public:
    // 需要计算的数据
    int _x;
    int _y;
    // 需要计算的种类
    char _op; //+ - * / %
};

// 定制的响应2
class Response
{
public:
    Response() : _exitCode(0), _result(0)
    {
    }
    ~Response()
    {
    }
    // 序列化 -- 不仅仅是在网络中应用,本地也是可以直接使用的
    //  "_exitCode _result"
    void serialize(std::string *out)
    {
#ifdef MY_SELF
        string ec = to_string(_exitCode);
        string res = to_string(_result);

        *out = ec + SPACE + res;
#else
        //json
        Json::Value root;
        root["code"]=_exitCode;
        root["res"]=_result;
        Json::FastWriter fw;
        // Json::StyledWriter fw;
        *out = fw.write(root);
#endif
    }
    // 反序列化
    bool deserialize(const std::string &in)
    {
#ifdef MY_SELF
        size_t pos = in.find(SPACE);
        if (pos == string::npos)
            return false;
        string codeStr = in.substr(0, pos);
        string resStr = in.substr(pos + SPACE_LEN);
        // 将反序列化的结果写入到内部成员中,形成结构化数据.
        _exitCode = stoi(codeStr);
        _result = stoi(resStr);
        return true;
#else
        //json
        Json::Value root;
        Json::Reader rd;
        rd.parse(in,root);
        _exitCode = root["code"].asInt();
        _result = root["res"].asInt();
        return true;
#endif
    }
    void debug()
    {
        cout << "#####################################" << endl;
        printf("_exitCode=%d,_result=%d\n", _exitCode, _result);
        cout << "#####################################" << endl;
    }

public:
    // 退出状态，0标识运算结果合法，非0标识运算结果是非法的，！0是几就表示是什么原因错了！
    int _exitCode;

    int _result;
};

bool makeRequest(string &str, Request *req)
{
    char strtmp[BUFFER_SIZE];
    snprintf(strtmp, sizeof strtmp, "%s", str.c_str());
    char *left = strtok(strtmp, OPS);
    if (!left)
        return false;
    char *right = strtok(nullptr, OPS);
    if (!right)
        return false;

    char mid = str[strlen(left)];
    req->_x = stoi(left);
    req->_y = stoi(right);
    req->_op = mid;
    return true;
}

clientTcp.cc

#include "util.hpp"
#include"Protocol.hpp"
#include "log.hpp"

using namespace std;

volatile bool quit =false;

static void Usage(string proc)
{
    cerr << "Usage\n\t" << proc << " ip port" << endl;
    cerr << "Example\n\t" << proc << " 127.0.0.1 8080\n"
         << endl;
}

// 2.需要bind吗??--->需要，但是不需要显式的bind！
// 3.需要listen吗？不需要的！
// 4.需要accept吗？不需要的！
// ./clientTcp serverIp serverPort
int main(int argc, char *argv[])
{
    if (argc != 2 && argc != 3)
    {
        Usage(argv[0]);
        exit(USAGE_ERR);
    }
    string serverIp = argv[1];
    uint16_t serverPort = stoi(argv[2]);

    // 1.创建套接字
    int sock = socket(AF_INET, SOCK_STREAM, 0);
    if (sock < 0)
    {
        cerr << "socket: " << strerror(errno) << endl;
        exit(SOCKET_ERR);
    }

    // 2.connect,发起连接请求，你想谁发起请求呢？当然是想服务器发起请求喽
    // 2.1 先填充需要连接的远端主机的基本信息
    struct sockaddr_in server;
    memset(&server, 0, sizeof(server));
    server.sin_family = AF_INET;
    inet_aton(serverIp.c_str(), &server.sin_addr);
    server.sin_port = ntohs(serverPort);
    // 2.2发送请求，connect会自动帮我们进行bind！
    if (connect(sock, (const sockaddr *)&server, sizeof(server)) != 0)
    {
        cerr << "connect: " << strerror(errno) << endl;
        exit(CONN_ERR);
    }
    cout << "info connect success: " << sock << endl;
    //trimStr(message); //对多输入的空格进行清洗
    string message;
    while(!quit)
    {
        cout<<"请输入表达式>> "; // 1 + 1
        getline(cin,message);
        if(strcasecmp(message.c_str(),"quit")==0)
        {    
            quit=true;
            continue;
        }     
        Request req;
        if(!makeRequest(message,&req))
            continue;
        std::string package;
        req.serialize(&package);
        cout<<"debug->serialize-> \n"<<package<<endl;
        package = encode(package,package.size());
        cout<<"debug->encode-> \n"<<package<<endl;
        ssize_t s=write(sock,package.c_str(),package.size());
        if(s>0)
        {
            char buffer[BUFFER_SIZE];
            size_t s =read(sock,buffer,sizeof(buffer)-1);
            if(s > 0) buffer[s]=0;
            cout<<buffer<<endl;
            string echoPackage = buffer;
            Response resp;
            uint32_t len=0;
            cout<<"debug->get response-> "<<echoPackage<<endl;
            string tmp = decode(echoPackage,&len);
            if(len > 0)
            {
                echoPackage = tmp;
                cout<<"debug->decode-> "<<echoPackage<<endl;
                resp.deserialize(echoPackage);
                printf("[exitcode: %d] %d\n",resp._exitCode,resp._result);
            }
        }
        else if(s<=0)
        {
            break;
        }
        message.clear();
    }

    close(sock);
    return 0;
}

calServer.cc

#include "util.hpp"
#include "log.hpp"
#include "ThreadPool.hpp"
#include "Protocol.hpp"
#include "Task.hpp"
#include "daemonize.hpp"
using namespace std;

class ServerTcp;
struct ThreadData
{
    ThreadData(int sock, string clientIp, uint16_t clientPort, ServerTcp *ts)
        : _sock(sock), _clientIp(clientIp), _clientPort(clientPort), _this(ts)
    {
    }
    int _sock;
    string _clientIp;
    uint16_t _clientPort;
    ServerTcp *_this;
};
static Response calcaulator(const Request &req)
{
    Response resp;
    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._exitCode = 1; // 1:除0
            else
                resp._result = req._x / req._y;
        }
        break;
    case '%':
        {
            if(req._y == 0)
                resp._exitCode = 2; // 1:模0
            else
                resp._result = req._x % req._y;
        }
        break;
    default:
        resp._exitCode = 3; //非法操作符
        break;
    }
    return resp;
}

// 1. 全部手写
// 2.部分采用别人的方案 -- 序列化和反序列化的问题  -- xml,json,protobuf 
void netCal(int sock, const string &clientIp, uint16_t clientPort)
{
    assert(sock > 0);
    assert(!clientIp.empty());
    assert(clientPort > 1024);

    // 9\r\n100 + 200\r\n
    string inbuffer;
    while (true)
    {
        char buff[128];
        ssize_t s = read(sock, buff, sizeof(buff) - 1);
        if (s == 0)
        {
            logMessage(NOTICE, "client[%s:%d] close sock, service done", clientIp.c_str(), clientPort);
            break;
        }
        else if (s < 0)
        {
            logMessage(WARINING, "read client[%s:%d] error, errcode: %d, errormessage: %s",
                       clientIp.c_str(), clientPort, errno, strerror(errno));
            break;
        }
        cout<<buff<<endl;
        // read success
        buff[s] = 0;
        inbuffer += buff;
        // 1.检查inbuffer是不是已经具有了一个strPackage
        Request req;
        uint32_t packageLen = 0;
        string package = decode(inbuffer, &packageLen);
        if (packageLen == 0)
            continue; // 无法提取一个完整的报文,进行下一次读取
        // 2.证明已经获得一个完整的package
        if (req.deserialize(package))
        {
            req.debug();
            // 3.处理逻辑,输入一个req,得到一个resp
            Response resp = calcaulator(req); //resp是一个结构化的字段
            // 4.对resp进行序列化
            string respPackage;
            resp.serialize(&respPackage);
            // 5.对报文进行encode
            respPackage = encode(respPackage,respPackage.size());
            // 6.简单进行发送
            write(sock,respPackage.c_str(),respPackage.size());

        }
    }
}

class ServerTcp
{
public:
    ServerTcp(uint16_t port, string ip = "")
        : _listenSock(-1), _port(port), _ip(ip), _tp(nullptr)
    {
        _quit = false;
    }
    ~ServerTcp()
    {
    }

public:
    void init()
    {
        // 1.创建socket
        _listenSock = socket(AF_INET, SOCK_STREAM, 0);
        if (_listenSock < 0)
        {
            logMessage(FATAL, "socket:%s", strerror(errno));
            exit(SOCKET_ERR);
        }
        logMessage(DEBUG, "socket:%s,%d", strerror(errno), _listenSock);
        // 2.bind绑定
        // 2.1填充服务器
        struct sockaddr_in local; // 用户栈
        memset(&local, 0, sizeof local);
        local.sin_family = AF_INET;
        local.sin_port = htons(_port);
        _ip.empty() ? (local.sin_addr.s_addr = INADDR_ANY) : (inet_aton(_ip.c_str(), &local.sin_addr));
        // 2.2本地socket信息，写入_sock对应的内核区域
        if (bind(_listenSock, (const sockaddr *)&local, sizeof local) < 0)
        {
            logMessage(FATAL, "bind: %s", strerror(errno));
            exit(BIND_ERR);
        }
        logMessage(DEBUG, "bind: %s", strerror(errno));
        // 3.监听socket，为何要监听呢？tcp是面向连接的！
        if (listen(_listenSock, 5 /*后面再说*/) < 0)
        {
            logMessage(FATAL, "listen: %s", strerror(errno));
            exit(LISTEN_ERR);
        }
        logMessage(DEBUG, "listen: %s", strerror(errno));
        // 允许别人来连接你了

        // 4.加载线程池
        _tp = ThreadPool<Task>::getInstance();
    }
    void loop()
    {
        // signal(SIGCHLD,SIG_IGN);//只在Linux下有效
        _tp->start();
        logMessage(DEBUG, "thread pool start success,thread num: %d", _tp->ThreadNum());

        while (!_quit)
        {
            struct sockaddr_in peer;
            socklen_t len = sizeof(peer);
            // 4.获取连接,accept的返回值是一个新的socket fd？？
            // 4.1 _listenScok:监听&&获取新的连接--->sock
            // 4.2 serviceSock:给用户提供新的socket服务
            int serviceSock = accept(_listenSock, (struct sockaddr *)&peer, &len);
            if (_quit)
                break;
            if (serviceSock < 0)
            {
                // 获取连接失败
                logMessage(WARINING, "accept: &s[%d]", strerror(errno), serviceSock);
                continue;
            }
            // 4.1获取客户端基本信息
            uint16_t peerPort = ntohs(peer.sin_port);
            string peerIp = inet_ntoa(peer.sin_addr);
            logMessage(DEBUG, "accept: %s | %s[%d],socker fd: %d",
                       strerror(errno), peerIp.c_str(), peerPort, serviceSock);

            // 5.提供服务，小写转大写
            Task t(serviceSock, peerIp, peerPort, netCal);
            _tp->push(t);

            // logMessage(DEBUG,"server provide service start ...");
            // sleep(1);
        }
    }

    bool quitServer()
    {
        _quit = true;
    }

private:
    int _listenSock;
    uint16_t _port;
    string _ip;
    // 引入线程池
    ThreadPool<Task> *_tp;
    // 安全退出
    bool _quit;
};
static void Usage(string proc)
{
    cerr << "Usage\n\t" << proc << " port ip" << endl;
    cerr << "Example\n\t" << proc << " 8080  127.0.0.1\n"
         << endl;
}

ServerTcp *svrp = nullptr;

void sighandler(int signo)
{
    if (signo == 3 && svrp != nullptr)
    {
        svrp->quitServer();
    }
    logMessage(DEBUG, "server quit save!");
}

// ./serverTcp local_port [local_ip]
int main(int argc, char *argv[])
{
    if (argc != 2 && argc != 3)
    {
        Usage(argv[0]);
        exit(USAGE_ERR);
    }
    uint16_t port = stoi(argv[1]);
    string ip;
    if (argc == 3)
    {
        ip = argv[2];
    }
    signal(3, sighandler);
    ServerTcp svr(port, ip);
    svr.init();
    svrp = &svr;
    svr.loop();
    return 0;
}

无论我们采用手写版本还是第三方库版本,又或者是其他的方案,只要保证,一段发送时构造的数据,在另一端能够正确的进行解析,这个方案就是可行的! 这种约定,就是应用层协议!

总结

(本小节完!)

秒客网

[Linux理论基础1]----手写和使用json完成[序列化和反序列化]

文章目录

前言

一、应用层

二、再谈"协议"

三、网络版计算器

手写版本

使用第三方库json实现

完整代码

总结

相关文章

[Linux理论基础1]----手写和使用json完成[序列化和反序列化]

文章目录

前言

一、应用层

二、再谈"协议"

三、 网络版计算器

手写版本

使用第三方库json实现

完整代码

总结

相关文章

三、网络版计算器