自定义协议、序列化与反序列化

在编写TCP和UDP程序的时候，我们很自然的就使用了读取的函数对数据进行获取，对于UDP来说提供的是无连接的以数据报的形式进行传输，对于TCP来说是面向数据流的，在之前的程序中我们只是进行了读取的操作，但是并没有对读取的内容进行分析。那如果我们要传输一些结构化的数据的话，那么就需要引入"协议"这个概念。

网络版计算器

在本文中将实现一个服务器版本的加法器，需要客户端把要计算的两个加数发过去,然后由服务器进行计算, 最后再把结果返回给客户端。

协议的约定

我们在这里约定信息，让协议能够更好的进行实现：

• 客户端发送一个形如"1+1"的字符串；
• 这个字符串中有两个操作数, 都是整形；
• 两个数字之间会有一个字符是运算符, 运算符只能是 “±*/%”；
• 数字和运算符之间没有空格；

序列化与反序列化

• 定义结构体来表示我们需要交互的信息；
• 发送数据时将这个结构体按照一个规则转换成字符串, 接收到数据的时候再按照相同的规则把字符串转化回结构体；
• 同时通过使用Jsoncpp的库进行协议的序列化与反序列化

TcpServer.hpp

namespace tcpserver_ns{
    using namespace protocol_ns; // 使用自定义协议的工作空间
    class TcpServer;
    using func_t = std::function<Response (const Request&)>;    
    class ThreadData{
    public:
        ThreadData(int sock, std::string ip, uint16_t port, TcpServer *tsvrp)
            : _sock(sock) , _ip(ip) ,_port(port), _tsvrp(tsvrp)
        {}
        ~ThreadData()
        {}
    public:
        int _sock;
        std::string _ip;
        uint16_t _port;
        TcpServer* _tsvrp;
    };
    class TcpServer{
    public:
        TcpServer(func_t func, uint16_t port) : _func(func), _port(port){}
        void InitServer(){ // 初始化相关的套接字信息
            _listensock.Socket();
            _listensock.Bind(_port);
            _listensock.Listen();
            logMessage(Info, "init server done, listensock: %d", _listensock.Fd());
        }
        static void* ThreadRoutine(void* args){
            pthread_detach(pthread_self());
            logMessage(Debug, "thread running ...");
            ThreadData* td = static_cast<ThreadData*>(args);
            td->_tsvrp->ServiceIO(td->_sock, td->_ip, td->_port); // 调用ServiceIO进行数据的读取与写入
            logMessage(Debug, "thread quit, client quit ... ");
            delete td;
            return nullptr;
        }
        void Start(){ // Accept 建立连接
            for(;;){
                std::string clientip;
                uint16_t clientport;
                int sock = _listensock.Accept(&clientip, &clientport);
                if (sock < 0) continue;
                logMessage(Debug, "get a new client, client info : [%s:%d]", clientip.c_str(), clientport);
                pthread_t tid;
                ThreadData *td = new ThreadData(sock, clientip, clientport, this); // 构建线程的数据信息
                pthread_create(&tid, nullptr, ThreadRoutine, td); // 创建线程运行ThreadRoutine
            }

        } 
        // 这个函数是被多线程调用的
        // 这里如果我们直接使用前面的文章中使用的read函数进行读取就无法保证获取的数据是我们想要的与计算相关的信息，
        // 因此此时需要我们自己根据自定义的协议来处理数据 -- 在这里我们规定每次我们需要的完整的报文是 "7"\r\n"10 + 20"\r\n 这样的形式，前面的数字表示有效报文的长度，报文长度与有效载荷之间通过"\r\n"来隔开
        void ServiceIO(int sock, const std::string &ip, const uint16_t &port){
            std::string inbuffer; // 放在外面，防止每次循环被释放
            while(true){
                // 0. 怎么保证读到的是一个完整的字符串报文？ "7"\r\n""10 + 20"\r\n
                std::string package;
                int n = ReadPackage(sock, inbuffer, &package); // 对获取的数据流进行处理，分出协议所需要的报文
                if (n == -1)
                    break;
                else if (n == 0)
                    continue;
                else{
                    // 一定得到了一个 "7"\r\n""10 + 20"\r\n
                    // 1. 你需要的只是有效载荷 "10 + 20"
                    package = RemoveHeader(package, n); // 对报文的有效载荷进行分离，获取有效载荷
                    // decode
                    // 2. 假设已经读到了一个完整的string "10 + 20"
                    Request req;
                    req.Deserialize(package); // 对读到的request字符串要进行反序列化
                    // 3. 直接提取用户的请求数据啦
                    Response resp = _func(req); // 业务逻辑！处理响应的计算业务
                    // 4. 给用户返回响应 - 序列化
                    std::string send_string;
                    resp.Serialize(&send_string); // 对计算完毕的response结构要进行序列化，形成可发送字符串
                    // 5. 添加报头
                    send_string = AddHeader(send_string); // 给需要返回的结果添加报头
                    //encode
                    // 6. 发送到网络 -- 弱化
                    send(sock, send_string.c_str(), send_string.size(), 0); // 简易版本的发送
                }
            }
            close(sock);
        }
        ~TcpServer(){
            _listensock.Close();
        }
    private:
        uint16_t _port;
        Sock _listensock;
        func_t _func;
    };
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

Protocol.hpp

自定义协议类以及Jsoncpp库使用：

#include 
#include "Util.hpp"
#define MYSELF 1
namespace protocol_ns{
	#define SEP " "
	#define SEP_LEN strlen(SEP)
	#define HEADER_SEP "\r\n"
	#define HEADER_SEP_LEN strlen(HEADER_SEP)

    // "长度"\r\n""_x _op _y"\r\n
    // "10 + 20" => "7"\r\n""10 + 20"\r\n => 报头 + 有效载荷
    // 请求/响应 = 报头\r\n有效载荷\r\n
    std::string AddHeader(const std::string &str){ // 添加报头
        std::cout << "AddHeader 之前:\n" << str << std::endl;
        std::string s = std::to_string(str.size());
        s += HEADER_SEP;
        s += str;
        s += HEADER_SEP;
        std::cout << "AddHeader 之后:\n" << s << std::endl;
        return s;
    }

    // "7"\r\n""10 + 20"\r\n => "10 + 20"
    std::string RemoveHeader(const std::string &str, const int &len){ // 移除报头
        std::cout << "RemoveHeader 之前:\n" << str << std::endl;
        std::string res = str.substr(str.size() - HEADER_SEP_LEN - len, len);
        std::cout << "RemoveHeader 之后:\n" << res << std::endl;
        return res;
    }
    int ReadPackage(int sock, std::string &inbuffer, std::string *package){ // 正确读取需要的报文
        std::cout << "ReadPackage inbuffer 之前:\n" << inbuffer << std::endl;
        // 边读取
        char buffer[1024];
        ssize_t s = recv(sock, buffer, sizeof(buffer) - 1, 0);
        if (s <= 0)
            return -1; // 读取出错
        buffer[s] = 0;
        inbuffer += buffer;
        std::cout << "ReadPackage inbuffer 之中:\n" << inbuffer << std::endl;
        // 边分析
        auto pos = inbuffer.find(HEADER_SEP);
        if (pos == std::string::npos)
            return 0;                                                    // 读取的不完善
        std::string lenStr = inbuffer.substr(0, pos);                    // 获取了头部字符串
        int len = Util::toInt(lenStr);                                   // "123" -> 123
        int targetpackageLen = lenStr.size() + len + 2 * HEADER_SEP_LEN; // 添加了报头的目标字符串长度
        if (inbuffer.size() < targetpackageLen)
            return 0;                                    // 读取的不完善
        *package = inbuffer.substr(0, targetpackageLen); // 提取报文
        // 到此为止，inbuffer 什么都没动
        inbuffer.erase(0, targetpackageLen); // 从inbuffer中直接移除整个报文
        std::cout << "ReadPackage inbuffer 之后:\n" << inbuffer << std::endl;
        return len; // 读取成功
    }
    // Request && Response都要提供序列化和反序列化功能
    class Request{
    public:
        Request()
        {}
        Request(int x, int y, char op) : _x(x), _y(y), _op(op)
        {}
        // 目前 "_x _op _y"
        bool Serialize(std::string *outStr){
            *outStr = "";
#ifdef MYSELF
            std::string x_string = std::to_string(_x);
            std::string y_string = std::to_string(_y);
            *outStr = x_string + SEP + _op + SEP + y_string;
            std::cout << "Request Serialize:\n" << *outStr << std::endl;
#else
            Json::Value root; // Value：一种万能对象，接受任意的kv对象
            root["x"] = _x;
            root["y"] = _y;
            root["op"] = _op; // 填充字段
            // Json::FastWriter writer; // Writer：是用来进行序列化的 struct -> string
            Json::StyledWriter writer; // 将Json转成好看一点的字符串
            // {
            //     "op" : 43,
            //     "x" : 1,
            //     "y" : 1
            // }
            *outStr = writer.write(root);
#endif 
            return true;
        }
        bool Deserialize(const std::string &inStr){
#ifdef MYSELF
            // inStr : 10 + 20 => [0]=>10, [1]=>+, [2]=>20
            // string -> vector
            std::vector<std::string> result;
            Util::StringSplit(inStr, SEP, &result);
            if (result.size() != 3)
                return false;
            if (result[1].size() != 1)
                return false;
            _x = Util::toInt(result[0]);
            _y = Util::toInt(result[2]);
            _op = result[1][0];
#else
            Json::Value root;
            Json::Reader reader; // Reader：用来进行反序列化的
            reader.parse(inStr, root);
            _x = root["x"].asInt();
            _y = root["y"].asInt();
            _op = root["op"].asInt();
#endif
            return true;
        }
        ~Request()
        {}
    public:
        // _x + _op + _y
        int _x;
        int _y;
        char _op;
    };

    class Response{
    public:
        Response()
        {}
        Response(int result, int code) : _result(result), _code(code)
        {}
        bool Serialize(std::string *outStr){
            *outStr = "";
#ifdef MYSELF
            std::string res_string = std::to_string(_result);
            std::string code_string = std::to_string(_code);
            *outStr = res_string + SEP + code_string;
            std::cout << "Response Serialize:\n" << *outStr << std::endl;
#else
            Json::Value root;
            root["result"] = _result;
            root["code"] = _code;
            // Json::FastWriter writer;
            Json::StyledWriter writer;
            *outStr = writer.write(root);
#endif
            return true;
        }

        bool Deserialize(const std::string &inStr){
#ifdef MYSELF
            std::vector<std::string> result;
            Util::StringSplit(inStr, SEP, &result);
            if (result.size() != 2)
                return false;
            _result = Util::toInt(result[0]);
            _code = Util::toInt(result[1]);
#else
            Json::Value root;
            Json::Reader reader; // Reader：用来进行反序列化的
            reader.parse(inStr, root);
            _result = root["result"].asInt();
            _code = root["code"].asInt();
#endif
            return true;
        }
        ~Response()
        {}
    public:
        int _result;
        int _code; // 0->success 1,2,3,4代表不同的错误
    };
}

// Util.hpp 工具类的实现
class Util
{
public:
    // 输入: const &
    // 输出: *
    // 输入输出: &
    static bool StringSplit(const string &str, const string &sep, vector<string> *result){
        size_t start = 0;
        // + 20
        // "abcd efg" -> for(int i = 0; i < 10; i++) !=  for(int i = 0; i <= 9; i++)
        while (start < str.size()){
            auto pos = str.find(sep, start);
            if (pos == string::npos) break;
            result->push_back(str.substr(start, pos-start));
            // 位置的重新reload
            start = pos + sep.size();
        }
        if(start < str.size())  result->push_back(str.substr(start));
        return true;
    }
    static int toInt(const std::string &s){
        return atoi(s.c_str());
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190

CalculatorServer.cc

Response calculate(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 = 1;
        else
            resp._result = req._x / req._y;
        break;
    case '%':
        if (req._y == 0)
            resp._code = 2;
        else
            resp._result = req._x % req._y;
        break;
    default:
        resp._code = 3;
        break;
    }

    return resp;
}

int main(){
    uint16_t port = 8081;
    std::unique_ptr<TcpServer> tsvr(new TcpServer(calculate, port)); // TODO
    tsvr->InitServer();
    tsvr->Start();
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

CalculatorClient.cc

using namespace protocol_ns;
static void usage(std::string proc){
    std::cout << "Usage:\n\t" << proc << " serverip serverport\n"
              << std::endl;
}
enum{
    LEFT,
    OPER,
    RIGHT
};
// 10+20
Request ParseLine(const std::string &line){ // 将输入的数据通过ParseLine函数进行分割并添加到请求所需要的变量中
    std::string left, right;
    char op;
    int status = LEFT;
    int i = 0;
    while(i < line.size()){
        // if(isdigit(e)) left.push_back;

        switch (status){
        case LEFT:
            if (isdigit(line[i]))
                left.push_back(line[i++]);
            else
                status = OPER;
            break;
        case OPER:
            op = line[i++];
            status = RIGHT;
            break;
        case RIGHT:
            if (isdigit(line[i]))
                right.push_back(line[i++]);
            break;
        }
    }
    Request req;
    std::cout << "left: " << left << std::endl;
    std::cout << "right: " << right << std::endl;
    std::cout << "op: " << op << std::endl;
    req._x = std::stoi(left);
    req._y = std::stoi(right);
    req._op = op;
    return req;
}

// ./tcpclient serverip serverport
int main(int argc, char *argv[]){
    if (argc != 3){
        usage(argv[0]);
        exit(USAGE_ERR);
    }
    std::string serverip = argv[1];
    uint16_t serverport = atoi(argv[2]);
    Sock sock;
    sock.Socket();
    int n = sock.Connect(serverip, serverport);
    if (n != 0)
        return 1;
    std::string buffer;
    while (true){
        std::cout << "Enter# "; // 1+1,2*9 // 这里的目标是要将输入的数据构建成前面的形式
        std::string line;
        std::getline(std::cin, line);
        Request req = ParseLine(line);
        std::cout << "test: " << req._x << req._op << req._y << std::endl;
        // 1. 序列化
        std::string sendString;
        req.Serialize(&sendString);
        // 2. 添加报头
        sendString = AddHeader(sendString);
        // 3. send
        send(sock.Fd(), sendString.c_str(), sendString.size(), 0);
        // 4. 获取响应
        std::string package;
        int n = 0;
    START:
        n = ReadPackage(sock.Fd(), buffer, &package);
        if (n == 0)
            goto START;
        else if (n < 0)
            break;
        else
        {}
        // 5. 去掉报头
        package = RemoveHeader(package, n);
        // 6. 反序列化
        Response resp;
        resp.Deserialize(package);
        std::cout << "result: " << resp._result << "[code: " << resp._code << "]" << std::endl;
    }
    sock.Close();
    return 0;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94