应用层自定义协议以及序列化和反序列化

文章目录

应用层自定义协议以及序列化和反序列化
- 1、应用层自定义协议
- - 1.1、应用层
  - 1.2、协议
- 2、序列化和反序列化
- 3、TCP 为什么支持全双工
- 4、jsoncpp基础
- - 4.1、序列化
  - 4.2、反序列化
- 5、实现网络版计算器
- 6、手写序列化和反序列化

应用层自定义协议以及序列化和反序列化

1、应用层自定义协议

1.1、应用层

应用层是计算机网络体系结构中的最高层，也是物联网三层结构中的最顶层，它直接面向用户和应用程序提供服务。

1.2、协议

协议就是通信双方约定好的结构化的数据！（比如结构体）

官方解释：应用层协议定义了交换的报文类型（如请求报文和响应报文）、报文类型的语法（如报文中的各个字段及其详细描述）、字段的语义（即包含在字段中信息的含义），以及进程何时、如何发送报文及对报文进行响应的规则。

2、序列化和反序列化

通信双方在发送和接收数据的时候，一般有两种方案：

方案一：
直接使用结构体发送，比如发送一个成员变量内容为1+1的结构体。
class request{
private:
   int _x;		// 左操作数
   int _y;		// 右操作数
   char _oper; // 操作符
};
只需要填充相关的成员变量内容再发送即可。但是这个方案存在一个问题！
就是互联网上的两台主机可能字节序不一样，比如发送端是大端机，接收方是小端机，那么接受数据就得处理大小端的问题。
这个方案一般在底层的通信上使用。
方案二：

使用定义的结构体来表示我们要交互的信息
但是发送和接收数据的时候是使用字符串
即发送的时候把结构体的内容转成字符串（序列化），接收的时候把字符串转换成结构体（反序列化）！

当然了，不管是方案一还是方案二或者其他方案，只要保证一段发送的数据，另一方能够对数据进行正确的解析，就可以。这种约定就是应用层协议！

下面我们使用方案二来理解协议，我们自定义协议，并且对于序列化和反序列化，有现成的方案–jsoncpp。我们后面还会自己实现序列化和反序列化！

3、TCP 为什么支持全双工

在前面博客代码中我们有谈到read/recv可能会出现bug，为什么？

因为对于TCP协议，接收数据的时候可能因为网络或者其他原因，可能读到的数据不完全或者过多，那么读取数据就会出错。

下面用一张图来解释TCP的全双工：

还有就是接收缓冲区填满的情况，有些数据会丢失（当然，OS会处理好，使用滑动窗口，后面博客会讲）

在上图我们可以看到：

在任何一台主机上，TCP有两个缓冲区（发送和接收缓冲区），所以在内核中，可以在发消息的同时接收消息，即全双工！

这就是为什么一个文件描述符（sockfd）就能实现读写的原因！（可能还有一个疑问就是为什么一个文件描述符能指向两个缓冲区？其实就是一个缓冲区，通过某种手段给它分成了两个部分，后面博客会讲）。

实际数据什么时候发，发多少，出错了怎么办，由TCP协议控制（后面会详细解释传输层TCP协议），所以TCP叫做传输控制协议。

4、jsoncpp基础

jsoncpp是一个现成的序列化和反序列化的方案

Jsoncpp 是一个用于处理 JSON 数据的 C++ 库。它提供了将 JSON 数据序列化为字符串以及从字符串反序列化为 C++ 数据结构的功能。Jsoncpp 是开源的，广泛用于各种需要处理 JSON 数据的 C++ 项目中。

特性

简单易用：Jsoncpp 提供了直观的 API，使得处理 JSON 数据变得简单。

高性能：Jsoncpp 的性能经过优化，能够高效地处理大量 JSON 数据。

全面支持：支持 JSON 标准中的所有数据类型，包括对象、数组、字符串、数字、布尔值和 null。

错误处理：在解析 JSON 数据时，Jsoncpp 提供了详细的错误信息和位置，方便开发者调试。

当使用 Jsoncpp 库进行 JSON 的序列化和反序列化时，确实存在不同的做法和工具类可供选择。以下是对 Jsoncpp 中序列化和反序列化操作的详细介绍:

安装：
sudo apt-get install libjsoncpp-dev # Ubuntu
sudo yum install jsoncpp-devel # Centos

4.1、序列化

序列化指的是将数据结构或对象转换为一种格式，以便在网络上传输或存储到文件中。

这里就介绍两个：Json::FastWriter和Json::StyledWriter，用法相似。其中Json::FastWriter要更快，因为它不加额外的空格和换行符。

#include <iostream>
#include <fstream>
#include <string>
#include <jsoncpp/json/json.h>

struct stu
{
   int id;
   std::string name;
   double grade;
   char c;

   void DebugPrint()
   {
       std::cout << id << " " << name << " " << grade << " " << c << std::endl;
   }
};

int main()
{
   // 写
   Json::Value root;
   struct stu s = {1, "xp", 99.99, 'a'};
   // 序列化
   root["id"] = s.id;
   root["name"] = s.name;
   root["grade"] = s.grade;
   root["c"] = s.c;
   Json::FastWriter writer;
   // Json::StyledWriter writer; -- 一样使用，但速度更慢
   std::string str = writer.write(root);
   std::ofstream ofs("./text.txt");
   ofs << str;

   return 0;
}

命令行执行：g++ test.cc -ljsoncpp，也就是需要链接jsoncpp库，因为这不是内置的库。

运行结果：

4.2、反序列化

反序列化指的是将序列化后的数据重新转换为原来的数据结构或对象。Jsoncpp 提供：使用 Json::Reader方法进行反序列化，这里就介绍这个，因为简单。

#include <iostream>
#include <fstream>
#include <string>
#include <jsoncpp/json/json.h>

struct stu
{
   int id;
   std::string name;
   double grade;
   char c;

   void DebugPrint()
   {
       std::cout << id << " " << name << " " << grade << " " << c << std::endl;
   }
};

int main()
{
   // 读
   char buff[1024];
   std::ifstream ifs("./text.txt");
   if (!ifs.is_open())
   {
       return 1;
   }
   ifs.getline(buff, sizeof(buff) - 1);
   std::string res = buff;
   Json::Value root;
   struct stu s;
   Json::Reader reader;
   // 反序列化
   bool n = reader.parse(res, root);
   if (!n)
       return 2;
   s.id = root["id"].asInt();
   s.name = root["name"].asCString();
   s.grade = root["grade"].asDouble();
   s.c = root["c"].asInt(); // 没有asChar

   s.DebugPrint();
   return 0;
}

运行结果：注意，文件内容是Json::FastWriter格式

5、实现网络版计算器

这里我们对socket的API进行封装成Socket.hpp文件。

方便TCP服务可以调用，UDP服务也可以调用（这里没实现UDP的socket 的 API具体细节，可以模仿TCP自行实现）。

定制协议封装了Protocol.hpp文件

其中序列化是采用的jsoncpp的方案，但是序列化完的数据不能直接发，直接发可能会出现粘报问题（一次收到多个数据分离不了，或者一次收到的数据不完全，需要识别，等待收到至少一条完整的数据到达），因此封装了两个函数解析字符串来解决这个问题：Encode添加报头长度和分隔符，Decode是相反的功能。

比如：{"x":_x,"y":_y,"oper":_oper}这样发送可以吗？不行，不一定一次到达的数据刚好是1条，可能是半条，也可能是2条，因此我们需要对发送的数据进行封装："len\r\n{有效载荷}\r\n" – 其中len是有效载荷的长度。

Socket.hpp文件

#pragma once

#include <string.h>
#include <memory>

#include "Log.hpp"
#include "Comm.hpp"

namespace socket_ns
{
  const static int gbacklog = 8;

  class Socket;
  using socket_sptr = std::shared_ptr<Socket>; // 定义智能指针，以便于后面多态

 // 使用
  // std::unique_ptr<Socket> listensocket = std::make_unique<TcpSocket>();
  // listensocket->BuildListenSocket();
  // socket_sptr retsock = listensocket->Accepter();
  // retsock->Recv();
 // retsock->Send();

  // std::unique_ptr<Socket> clientsocket = std::make_unique<TcpSocket>();
  // clientsocket->BuildClientSocket();
  // clientsocket->Send();
  // clientsocket->Recv();

  class Socket
  {
 public:
      virtual void CreateSocketOrDie() = 0;
     virtual void BindSocketOrDie(InetAddr &addr) = 0;
      virtual void ListenSocketOrDie() = 0;
      virtual socket_sptr Accepter(InetAddr *addr) = 0;
      virtual bool Connector(InetAddr &addr) = 0;

      virtual int SockFd() = 0;

      virtual ssize_t Recv(std::string *out) = 0;
      virtual ssize_t Send(std::string &in) = 0;
      // virtual void Other() = 0;

  public:
     void BuildListenSocket(InetAddr &addr)
      {
          CreateSocketOrDie();
          BindSocketOrDie(addr);
          ListenSocketOrDie();
      }

     bool BuildClientSocket(InetAddr &addr)
      {
          CreateSocketOrDie();
          return Connector(addr);
      }
  };

  class TcpSocket : public Socket
  {
  public:
      TcpSocket(int sockfd = -1) : _socktfd(sockfd)
      {
      }
      virtual void CreateSocketOrDie() override
      {
          // 创建
          _socktfd = socket(AF_INET, SOCK_STREAM, 0); // 这个就是文件描述符
          if (_socktfd < 0)
          {
              LOG(FATAL, "create sockfd error, error code : %d, error string : %s", errno, strerror(errno));
              exit(CREATE_ERROR);
          }
          LOG(INFO, "create sockfd success");
      }
      virtual void BindSocketOrDie(InetAddr &addr) override
      {
          struct sockaddr_in local;
          bzero(&local, sizeof(local));
          local.sin_family = AF_INET;
          local.sin_port = htons(addr.Port());
          local.sin_addr.s_addr = INADDR_ANY;
          // 绑定
          int n = ::bind(_socktfd, CONV(&local), sizeof(local));
          if (n < 0)
          {
              LOG(FATAL, "bind sockfd error, error code : %d, error string : %s", errno, strerror(errno));
              exit(BIND_ERROR);
          }
          LOG(INFO, "bind sockfd success");
      }
      virtual void ListenSocketOrDie() override
      {
          // 监听
          int ret = ::listen(_socktfd, gbacklog);
          if (ret < 0)
          {
              LOG(FATAL, "listen error, error code : %d , error string : %s", errno, strerror(errno));
              exit(LISTEN_ERROR);
          }
          LOG(INFO, "listen success!");
      }
      virtual socket_sptr Accepter(InetAddr *addr) override
      {
          struct sockaddr_in peer;
          socklen_t len = sizeof(peer);
          // 获取新连接
          int newsockfd = accept(_socktfd, CONV(&peer), &len); // 建立连接成功，创建新文件描述符进行通信
          if (newsockfd < 0)
          {
              LOG(WARNING, "accept error, error code : %d , error string : %s", errno, strerror(errno));
              return nullptr;
         }
          LOG(INFO, "accept success! new sockfd : %d", newsockfd);
          *addr = peer;
          socket_sptr sock = std::make_shared<TcpSocket>(newsockfd); // 创建新的文件描述符，传出去以便于后面的Recv和Send
          return sock;
      }

      virtual bool Connector(InetAddr &addr) override
     {
          struct sockaddr_in local;
          bzero(&local, sizeof(local));
          local.sin_family = AF_INET;
          local.sin_port = htons(addr.Port());
          local.sin_addr.s_addr = inet_addr(addr.Ip().c_str());

          // 发起连接
          int n = ::connect(_socktfd, CONV(&local), sizeof(local));
          if (n < 0)
          {
             LOG(WARNING, "create connect error, error code : %d, error string : %s", errno, strerror(errno));
              return false;
          }
          LOG(INFO, "create connect success");
          return true;
     }

      virtual int SockFd() override
      {
          return _socktfd;
      }

      virtual ssize_t Recv(std::string *out) override
      {
          char buff[1024];
          ssize_t n = recv(_socktfd, buff, sizeof(buff) - 1, 0);
          if (n > 0)
          {
              buff[n] = 0;
              *out += buff; // 方便当数据到来不是刚好1条数据的时候，进行合并后来的数据
          }
          return n;
     }
      virtual ssize_t Send(std::string &in) override
      {
          ssize_t n = send(_socktfd, in.c_str(), in.size(),0);
         return n;
     }

 private:
     int _socktfd; // 用同一个_socket
 };
}

Calculate.hpp文件

#pragma once

#include <iostream>
#include <string>
#include "Protocol.hpp"

using namespace protocol_ns;

// 应用层
class Calculate
{
public:
  Calculate()
 {
  }

  std::unique_ptr<Response> Execute(const Request &req)
  {
      std::unique_ptr<Response> resptr = std::make_unique<Response>();

     switch (req._oper)
      {
      case '+':
          resptr->_result = req._x + req._y;
          resptr->_equation = std::to_string(req._x) + " " + req._oper + " " + std::to_string(req._y) + " = " + std::to_string(resptr->_result);
         break;

      case '-':
          resptr->_result = req._x - req._y;
          resptr->_equation = std::to_string(req._x) + " " + req._oper + " " + std::to_string(req._y) + " = " + std::to_string(resptr->_result);
          break;

      case '*':
          resptr->_result = req._x * req._y;
          resptr->_equation = std::to_string(req._x) + " " + req._oper + " " + std::to_string(req._y) + " = " + std::to_string(resptr->_result);
          break;
      case '/':
      {
          if (req._y == 0)
          {
              resptr->_flag = 1;
              resptr->_equation = "除0错误";
          }
          else
         {
              resptr->_result = req._x / req._y;
              resptr->_equation = std::to_string(req._x) + " " + req._oper + " " + std::to_string(req._y) + " = " + std::to_string(resptr->_result);
          }
          break;
      }

      case '%':
      {
          if (req._y == 0)
          {
              resptr->_flag = 2;
              resptr->_equation = "模0错误";
          }
          else
          {
              resptr->_result = req._x % req._y;
              resptr->_equation = std::to_string(req._x) + " " + req._oper + " " + std::to_string(req._y) + " = " + std::to_string(resptr->_result);
          }
          break;
      }
      default:
         resptr->_flag = 3;
         break;
     }
     return resptr;
 }
 ~Calculate() {}

private:
};

Comm.hpp文件

#pragma once
#include "InetAddr.hpp"

enum errorcode
{
  CREATE_ERROR = 1,
  BIND_ERROR,
 LISTEN_ERROR,
 SEND_ERROR,
 RECV_ERROR,
 CONNECT_ERROR,
 FORK_ERROR,
 USAGE_ERROR
};

#define CONV(ADDR) ((struct sockaddr *)ADDR)

std::string CombineIpAndPort(InetAddr addr)
{
 return "[" + addr.Ip() + ":" + std::to_string(addr.Port()) + "] ";
}

InetAddr.hpp文件

#pragma once

#include <iostream>
#include <sys/types.h> /* See NOTES */
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <string>

class InetAddr
{
 void GetAddress(std::string *ip, uint16_t *port)
 {
      // char *inet_ntoa(struct in_addr in);
      *ip = inet_ntoa(_addr.sin_addr);
      *port = ntohs(_addr.sin_port);
  }

public:
  InetAddr(const struct sockaddr_in &addr) : _addr(addr)
  {
      GetAddress(&_ip, &_port);
  }

  InetAddr(std::string ip, uint16_t port) : _ip(ip), _port(port)
 {
      _addr.sin_family = AF_INET;
      _addr.sin_port = htons(_port);
      _addr.sin_addr.s_addr = inet_addr(_ip.c_str());
  }
 InetAddr() {}

  std::string Ip()
  {
      return _ip;
  }

  uint16_t Port()
  {
     return _port;
  }
  bool operator==(InetAddr &addr)
  {
      return _ip == addr.Ip() && _port == addr.Port();
 }

 const struct sockaddr_in &GetAddr()
 {
      return _addr;
  }

 ~InetAddr() {}

private:
 struct sockaddr_in _addr;
 std::string _ip;
 uint16_t _port;
};

LockGuard.hpp文件

# pragma once

#include <pthread.h>


class LockGuard
{
public:
  LockGuard(pthread_mutex_t *mutex) : _mutex(mutex)
  {
     pthread_mutex_lock(_mutex); // 构造加锁
 }
  ~LockGuard()
 {
     pthread_mutex_unlock(_mutex); // 析构解锁
 }

private:
 pthread_mutex_t *_mutex;
};

Log.hpp文件

#pragma once

#include <string>
#include <iostream>
#include <fstream>
#include <unistd.h>
#include <stdarg.h>
#include <sys/types.h>
#include "LockGuard.hpp"

using namespace std;

bool isSave = false; // 默认向显示器打印
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
#define FILEPATH "./log.txt"

enum level
{
 DEBUG = 0,
 INFO,
 WARNING,
  ERROR,
  FATAL
};

void SaveToFile(const string &message)
{
 ofstream out(FILEPATH, ios_base::app);
 if (!out.is_open())
     return;
  out << message;
  out.close();
}

std::string LevelToString(int level)
{
  switch (level)
  {
  case DEBUG:
      return "Debug";
  case INFO:
      return "Info";
  case WARNING:
      return "Warning";
  case ERROR:
     return "Error";
 case FATAL:
     return "Fatal";
 default:
      return "Unknow";
  }
}

std::string GetTimeString()
{
  time_t curr_time = time(nullptr);
  struct tm *format_time = localtime(&curr_time);
  if (format_time == nullptr)
      return "None";
  char buff[1024];
  snprintf(buff, sizeof(buff), "%d-%d-%d %d:%d:%d",
           format_time->tm_year + 1900,
          format_time->tm_mon + 1,
          format_time->tm_mday,
          format_time->tm_hour,
          format_time->tm_min,
           format_time->tm_sec);
  return buff;
}

void LogMessage(const std::string filename, int line, bool issave, int level, const char *format, ...)
{
  std::string levelstr = LevelToString(level);
  std::string timestr = GetTimeString();
  pid_t pid = getpid();

 char buff[1024];
  va_list arg;
  // int vsnprintf(char *str, size_t size, const char *format, va_list ap); // 使用可变参数
  va_start(arg, format);
  vsnprintf(buff, sizeof(buff), format, arg);
  va_end(arg);

 LockGuard lock(&mutex);
 std::string message = "[" + timestr + "]" + "[" + levelstr + "]" + "[pid:" + std::to_string(pid) + "]" + "[" + filename + "]" + "[" + std::to_string(line) + "] " + buff + '\n';
 if (issave == false)
     std::cout << message;
  else
      SaveToFile(message);
}

// 固定文件名和行数
#define LOG(level, format, ...)                                               \
  do                                                                        \
  {                                                                         \
      LogMessage(__FILE__, __LINE__, isSave, level, format, ##__VA_ARGS__); \
  } while (0)

#define EnableScreen()  \
  do                  \
  {                   \
      isSave = false; \
  } while (0)

#define EnableFile()   \
 do                 \
  {                  \
      isSave = true; \
  } while (0)

void Test(int num, ...)
{
  va_list arg;
  va_start(arg, num);
  while (num--)
 {
     int data = va_arg(arg, int);
     std::cout << data << " ";
 }
 std::cout << std::endl;
 va_end(arg);
}

Main.cc文件

#include <iostream>
#include <memory>
#include "TcpServer.hpp"
#include "Protocol.hpp"
#include "Calculate.hpp"

using namespace protocol_ns;

using cal_t = std::function<std::unique_ptr<Response>(const Request &req)>;

void Usage()
{
 // printf("./udp_server serverip serverport\n");
  printf("Usage : ./udp_server serverport\n"); // ip 已经设置为0
}

class Service
{
public:
 Service(cal_t cb) : _cb(cb) {}
  void AddService(socket_sptr sockfd, InetAddr client)
  {
      // TCP是字节流（可以使用write和read接口），UDP是数据报
     std::string clientaddr = CombineIpAndPort(client);
      std::string recvmessage;

      while (true)
      {
          sleep(5);

          Request req; // 注意多线程问题，不能放在里面while

          // 1.接收数据
          int n = sockfd->Recv(&recvmessage);
          std::cout << "server recv:" << recvmessage << std::endl;

          if (n <= 0)
          {
              LOG(INFO, "client %s quit", clientaddr.c_str());
              break;
          }
          // 2.分析数据，确定完整报文
          std::string package;
         while (true)
          {
              package = Decode(recvmessage); // 可能为空
             if (package.empty())
                  break;
              cout << "after Decode recvmessage : " << recvmessage << std::endl;

              // 完整的一条有效数据
              std::cout << "server Decode:" << package << std::endl;

              // 3.反序列化
              req.DeSerialize(package); // 把_x,_y,_oper赋值

              // 4.业务处理
             std::unique_ptr<Response> resptr = _cb(req);

              // 5.序列化
              std::string sendmessage;
             resptr->Serialize(&sendmessage);
              std::cout << "server Serialize:" << sendmessage << std::endl;

              // 6.加上报头数据封装
              sendmessage = Encode(sendmessage);
              std::cout << "server Encode:" << sendmessage << std::endl;

             // 7.发送数据
             int n = sockfd->Send(sendmessage);
          }
     }
 }
 ~Service() {}

private:
  cal_t _cb;
};

int main(int argc, char *argv[])
{
 // if (argc != 3)
  if (argc != 2)
 {
      Usage();
      exit(USAGE_ERROR);
  }

  uint16_t serverport = std::stoi(argv[1]);

  // __nochdir = 1:在当前工作目录执行
  // __nochdir = 0:在根目录/工作目录执行
  // __noclose = 1:不进行重定向
  // __noclose = 0:进行重定向 /dev/null
 // int daemon(int __nochdir, int __noclose)
  // if(fork > 0) exit(0);
 // setsid();
  // 先创建子进程，再父进程退出，因为组长不能直接调用setsid();变成守护进程
  // daemon(0, 0);
  // 执行下面的代码不是当前进程，而是当前进程的子进程

  //EnableFile();

 Calculate cal; // 应用层
  cal_t servercal = std::bind(&Calculate::Execute, &cal, placeholders::_1);
 Service sev(servercal);
 service_t service = std::bind(&Service::AddService, &sev, placeholders::_1, placeholders::_2); // 表示层
 std::unique_ptr<TcpServer> tsvr = std::make_unique<TcpServer>(serverport, service);            //  会话层
 tsvr->Start();

 return 0;
}

Makefile文件

.PHONY:all
all:tcp_client tcp_server

tcp_client:TcpClient.cc
	g++ -o $@ $^ -std=c++14 -lpthread -ljsoncpp
tcp_server:Main.cc
	g++ -o $@ $^ -std=c++14 -lpthread -ljsoncpp

.PHONY:clean
clean:
	rm -f tcp_server tcp_client

Protocol.hpp文件

#pragma once

#include <string>
#include <jsoncpp/json/json.h>
#include <iostream>
#include <ctime>
#include <sys/types.h>
#include <unistd.h>


// 表示层
namespace protocol_ns
{
  const std::string SEP = "\r\n";
  const std::string CAL_SEP = " ";

  // 对发送数据进行封装
  // "len\r\n{有效载荷}\r\n" -- 其中len是有效载荷的长度
  std::string Encode(const std::string &inbuff)
  {
      int inbuff_len = inbuff.size();
      std::string newstr = std::to_string(inbuff_len);
      newstr += SEP;
      newstr += inbuff;
     newstr += SEP;
      return newstr;
  }

  // 解析字符串
  std::string Decode(std::string &outbuff)
  {
      int pos = outbuff.find(SEP);
      if (pos == std::string::npos)
      {
          // 没找到分隔符
          return std::string(); // 返回空串，等待接收到完整数据
      }
      // 找到分隔符
      std::string len_str = outbuff.substr(0, pos);
      if (len_str.empty())
          return std::string(); // 返回空串，等待接收到完整数据
      int data_len = std::stoi(len_str);
      // 判断长度是否符合要求
      int total_len = pos + SEP.size() * 2 + data_len; // 包装好的一条数据的长度
      if (outbuff.size() < total_len)
      {
          return std::string(); // 小于包装好的一条数据的长度，返回空串，等待接收到完整数据
      }
      // 大于等于包装好的一条数据的长度
      std::string message = outbuff.substr(pos + SEP.size(), data_len); // 有效数据
     outbuff.erase(0, total_len);                                      // 数据长度减少包装好的一条数据的长度，从前面开始移除
      return message;
  }

  class Request
  {
  public:
      Request() {}
      Request(int x, int y, char oper)
          : _x(x),
            _y(y),
           _oper(oper)
      {
      }

      // 序列化 -- 转化为字符串发送
      // {"x":_x,"y":_y,"oper":_oper}
      // 这样发送可以吗？不行，不一定一次到达的数据刚好是1条，可能是半条，也可能是2条，因此我们需要对发送的数据进行封装：
     // "len\r\n{有效载荷}\r\n" -- 其中len是有效载荷的长度
      void Serialize(std::string *out) // 要带出来
      {
          Json::Value root;
          root["x"] = _x;
         root["y"] = _y;
          root["oper"] = _oper;

          Json::FastWriter writer;
          std::string str = writer.write(root);

          *out = str;
      }

      // 反序列化 -- 解析
     bool DeSerialize(const std::string &in)
      {
         Json::Value root;
          Json::Reader reader;
          if (!reader.parse(in, root))
              return false;
         _x = root["x"].asInt();
          _y = root["y"].asInt();
          _oper = root["oper"].asInt();
          return true;
      }
      ~Request() {}

  public:
     int _x;
      int _y;
      char _oper; // +-*/% 如果不是这些操作法那就是非法的
  };

  class Response
  {
  public:
     Response() {}
      // 序列化 -- 转化为字符串发送
      void Serialize(std::string *out) // 要带出来
      {
          Json::Value root;
         root["result"] = _result;
         root["flag"] = _flag;
          root["equation"] = _equation;

          Json::FastWriter writer;
          std::string str = writer.write(root);

          *out = str;
      }

     // 反序列化 -- 解析
      bool DeSerialize(const std::string &in)
      {
         Json::Value root;
          Json::Reader reader;
          if (!reader.parse(in, root))
              return false;
          _result = root["result"].asInt();
          _flag = root["flag"].asInt();
         _equation = root["equation"].asString();
          return true;
      }
      ~Response() {}

  public:
      int _result = 0;
      int _flag = 0;                         // 0表示操作符正确，1表示除0错误，2表示取模0错误，3表示操作符错误
     string _equation = "操作符不符合要求"; // 等式
  };

  const std::string opers = "+-*/%&^";

 class CalFactory
  {
  public:
      CalFactory()
      {
         srand(time(nullptr) ^ getpid() ^ 2);
      }
      void Product(Request &req)
     {
          req._x = rand() & 5 + 1;
         usleep(req._x * 20);
          req._y = rand() % 10 + 5;
         // req._y = 0; // 测试
          usleep(req._x * req._y + 20);
          req._oper = opers[(rand() % opers.size())];
      }
     ~CalFactory() {}

 private:
  };
}

TcpClient.cc文件

#include <iostream>
#include <string>
#include <strings.h>
#include <unistd.h>

#include "Comm.hpp"
#include "Socket.hpp"
#include "Protocol.hpp"

using namespace socket_ns;
using namespace protocol_ns;

enum class Status
{
  NEW,
  CONNECTED,
  CONNECTING,
  DISCONNECTED,
  CLOSE
};

const int defaultsockfd = -1;
const int retryinterval = 1; // 重连间隔时间
const int retryamxtimes = 5; // 重连最大次数

class Connection
{
public:
  Connection(std::string serverip, uint16_t serverport)
      : _sockfdptr(std::make_unique<TcpSocket>()),
        _serverip(serverip),
       _serverport(serverport),
        _status(Status::NEW),
       _retry_interval(retryinterval),
        _retry_max_times(retryamxtimes)
  {
  }

  Status ConnectStatus()
  {
      return _status;
  }

  void Connect()
  {
      InetAddr server(_serverip, _serverport);
      bool ret = _sockfdptr->BuildClientSocket(server);
      if (!ret)
      {
          DisConnect(); //
          _status = Status::DISCONNECTED;
         return;
      }
      std::cout << "connect success" << std::endl;
     _status = Status::CONNECTED; // 已连接
 }

 void Process()
 {
     while (true)
     {
         // std::cout << "Please Enter#  ";
         // std::string sendmessage;// 不是{"": ,}类型
         // std::getline(std::cin, sendmessage);

         sleep(1);
         Request req;
         CalFactory cal;
         // 1.对需要发送的数据进行序列化
         std::string sendmessage;

         // 1.1.这里一下构建5个请求并放在一起
         // for (int i = 0; i < 5; ++i)
         // {
          //     cal.Product(req);
          //     std::string sub_sendstr;
          //     req.Serialize(&sub_sendstr);
          //     std::cout << "client Serialize:" << sub_sendstr << std::endl;
          //     // 2.对序列化后的数据进行加报头等打包
         //     sub_sendstr = Encode(sub_sendstr);
         //     std::cout << "client Encode:" << sub_sendstr << std::endl;
         //     sendmessage += sub_sendstr;
         // }

         cal.Product(req);
         req.Serialize(&sendmessage);
         std::cout << "client Serialize:" << sendmessage << std::endl;
         // 2.对序列化后的数据进行加报头等打包
          sendmessage = Encode(sendmessage);
          std::cout << "client Encode:" << sendmessage << std::endl;

          // std::cout << "sendmessage : " << sendmessage << std::endl;
          // 3.发送数据
          int n = _sockfdptr->Send(sendmessage);
          if (n < 0)
          {
              _status = Status::CLOSE; // 发送不成功就退出
             LOG(FATAL, "send error, errno : %d ,error string : %s", errno, strerror(errno));
              break;
          }
          // 发送成功
          std::string recvmessage;
         // 4.接收数据
          int m = _sockfdptr->Recv(&recvmessage);
          if (m <= 0)
          {
              _status = Status::DISCONNECTED; // 接收不成功就重连
              std::cerr << "recv error" << std::endl;
              break;
          }
          // 接收成功
          // 5.分析数据，确定完整报文
          std::string package;
          while (true)
          {
              package = Decode(recvmessage); // 可能为空
             if (package.empty())
                  break;
              // 完整的一条有效数据
              Response resp;
              // 6.反序列化
              resp.DeSerialize(package); // 把_result,_flag赋值
              // 7.处理返回数据
              std::cout << "Server Echo$ " << "result : " << resp._result << " , flag :" << resp._flag << " --- equation : " << resp._equation << std::endl;
         }
      }
  }

  void ReConnect()
  {
     _status = Status::CONNECTING;
      int cnt = 1;
      while (true)
      {
          Connect();
          if (_status == Status::CONNECTED)
          {
              break;
          }
          std::cout << "正在重连，重连次数 : " << cnt++ << std::endl;
          if (cnt > _retry_max_times)
          {
              _status = Status::CLOSE; // 重连失败
             std::cout << "重连失败，请检查网络.." << std::endl;
              break;
          }
          sleep(_retry_interval);
      }
  }

 void DisConnect()
  {
      if (_sockfdptr->SockFd() > defaultsockfd)
      {
          close(_sockfdptr->SockFd());
      }
  }

private:
  std::unique_ptr<Socket> _sockfdptr;
  std::string _serverip;
  uint16_t _serverport;
  Status _status;
  int _retry_interval;
  int _retry_max_times;
};

class TcpClient
{
public:
  TcpClient(std::string serverip, uint16_t serverport) : _connect(serverip, serverport)
  {
  }
  void Execute()
  {
      while (true)
      {
          switch (_connect.ConnectStatus())
          {
          case Status::NEW:
              _connect.Connect();
              break;
          case Status::CONNECTED:
              _connect.Process();
              break;
          case Status::DISCONNECTED:
              _connect.ReConnect();
             break;
          case Status::CLOSE:
              _connect.DisConnect();
              return; // 断开连接了，重连不管用了
          default:
              break;
          }
      }
  }

private:
  Connection _connect;
};

void Usage()
{
  std::cout << "Please use format : ./tcp_client serverip serverport" << std::endl;
}

int main(int argc, char *argv[])
{
  if (argc != 3)
 {
      Usage();
      exit(USAGE_ERROR);
  }

  std::string serverip = argv[1];
  uint16_t serverport = std::stoi(argv[2]);

 TcpClient tcpclient(serverip, serverport);
 tcpclient.Execute();
  return 0;
}

TcpServer.hpp文件

#pragma once

#include <sys/types.h> /* See NOTES */
#include <sys/wait.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>

#include <error.h>
#include <string.h>
#include <pthread.h>
#include <functional>

#include "Log.hpp"
#include "InetAddr.hpp"
#include "Comm.hpp"
#include "Socket.hpp"

using namespace socket_ns;

using service_t = std::function<void(socket_sptr sockfd, InetAddr client)>;

// 会话层
// 声明
class TcpServer;

class ThreadData
{
public:
  ThreadData(socket_sptr sockfd, InetAddr addr, TcpServer *self)
     : _sockfd(sockfd), _addr(addr), _self(self) {}
 ~ThreadData() = default;

public:
 socket_sptr _sockfd;
 InetAddr _addr;
 TcpServer *_self;
};

class TcpServer
{
public:
  TcpServer(uint16_t port, service_t service)
      : _localaddr("0", port),
        _listensock(std::make_unique<TcpSocket>()),
        _service(service),
        _isrunning(false)
 {
      _listensock->BuildListenSocket(_localaddr);
  }

  static void *HandlerService(void *args)
  {
      pthread_detach(pthread_self()); // 分离线程
     ThreadData *td = static_cast<ThreadData *>(args);
     td->_self->_service(td->_sockfd, td->_addr);
     ::close(td->_sockfd->SockFd()); // 服务结束，关闭文件描述符，避免文件描述符泄漏
     delete td;
     return nullptr;
 }

 void Start()
 {
     _isrunning = true;
     while (_isrunning)
     {
         InetAddr peerAddr;
         socket_sptr normalsock = _listensock->Accepter(&peerAddr);

         // v2 -- 多线程
         pthread_t tid;
         ThreadData *td = new ThreadData(normalsock, peerAddr, this); // 传指针
         pthread_create(&tid, nullptr, HandlerService, td);           // 这里创建线程后，线程去做执行任务，主线程继续向下执行 , 并且线程不能关闭sockf，线程和进程共享文件描述符表
     }
     _isrunning = false;
 }

 ~TcpServer()
 {
 }

private:
 service_t _service;
 InetAddr _localaddr;
 std::unique_ptr<Socket> _listensock;
 bool _isrunning;
};

运行结果：

6、手写序列化和反序列化

这里我们只需要修改Protocol.hpp文件

#pragma once

#include <string>
#include <jsoncpp/json/json.h>
#include <iostream>
#include <ctime>
#include <sys/types.h>
#include <unistd.h>

// #define SELF 1; // SELF=1就用自定义的序列化和反序列化，否则用默认的

// 表示层
namespace protocol_ns
{
   const std::string SEP = "\r\n";
   const std::string CAL_SEP = " ";

   // 对发送数据进行封装
   // "len\r\n{有效载荷}\r\n" -- 其中len是有效载荷的长度
   std::string Encode(const std::string &inbuff)
   {
       int inbuff_len = inbuff.size();
       std::string newstr = std::to_string(inbuff_len);
       newstr += SEP;
       newstr += inbuff;
       newstr += SEP;
       return newstr;
   }

   // 解析字符串
   std::string Decode(std::string &outbuff)
   {
       int pos = outbuff.find(SEP);
       if (pos == std::string::npos)
       {
           // 没找到分隔符
           return std::string(); // 返回空串，等待接收到完整数据
       }
       // 找到分隔符
       std::string len_str = outbuff.substr(0, pos);
       if (len_str.empty())
           return std::string(); // 返回空串，等待接收到完整数据
       int data_len = std::stoi(len_str);
       // 判断长度是否符合要求
       int total_len = pos + SEP.size() * 2 + data_len; // 包装好的一条数据的长度
       if (outbuff.size() < total_len)
       {
           return std::string(); // 小于包装好的一条数据的长度，返回空串，等待接收到完整数据
       }
       // 大于等于包装好的一条数据的长度
       std::string message = outbuff.substr(pos + SEP.size(), data_len); // 有效数据
       outbuff.erase(0, total_len);                                      // 数据长度减少包装好的一条数据的长度，从前面开始移除
       return message;
   }

   class Request
   {
   public:
       Request() {}
       Request(int x, int y, char oper)
           : _x(x),
             _y(y),
             _oper(oper)
       {
       }

       // 序列化 -- 转化为字符串发送
       // {"x":_x,"y":_y,"oper":_oper}
       // 这样发送可以吗？不行，不一定一次到达的数据刚好是1条，可能是半条，也可能是2条，因此我们需要对发送的数据进行封装：
       // "len\r\n{有效载荷}\r\n" -- 其中len是有效载荷的长度
       void Serialize(std::string *out) // 要带出来
       {
#ifdef SELF
           // "len\r\nx op y\r\n" -- 自定义序列化和反序列化
           std::string data_x = std::to_string(_x);
           std::string data_y = std::to_string(_y);
           *out = data_x + CAL_SEP + _oper + CAL_SEP + data_y;
#else
           Json::Value root;
           root["x"] = _x;
           root["y"] = _y;
           root["oper"] = _oper;

           Json::FastWriter writer;
           std::string str = writer.write(root);

           *out = str;
#endif
       }

       // 反序列化 -- 解析
       bool DeSerialize(const std::string &in)
       {
#ifdef SELF
           auto left_blank_pos = in.find(CAL_SEP);
           if (left_blank_pos == std::string::npos)
               return false;
           std::string x_str = in.substr(0, left_blank_pos);
           if (x_str.empty())
               return false;
           auto right_blank_pos = in.rfind(CAL_SEP);

           if (right_blank_pos == std::string::npos)
               return false;
           std::string y_str = in.substr(right_blank_pos + 1);
           if (y_str.empty())
               return false;
           if (left_blank_pos + 1 + CAL_SEP.size() != right_blank_pos)
               return false;

           _x = std::stoi(x_str);
           _y = std::stoi(y_str);
           _oper = in[right_blank_pos - 1];
           return true;

#else
           Json::Value root;
           Json::Reader reader;
           if (!reader.parse(in, root))
               return false;
           _x = root["x"].asInt();
           _y = root["y"].asInt();
           _oper = root["oper"].asInt();
           return true;
#endif
       }
       ~Request() {}

   public:
       int _x;
       int _y;
       char _oper; // +-*/% 如果不是这些操作法那就是非法的
   };

   class Response
   {
   public:
       Response() {}
       // 序列化 -- 转化为字符串发送
       void Serialize(std::string *out) // 要带出来
       {
#ifdef SELF
           // "len\r\nresult flag equation\r\n"
           std::string data_res = std::to_string(_result);
           std::string data_flag = std::to_string(_flag);
           *out = data_res + CAL_SEP + data_flag + CAL_SEP + _equation;
#else
           Json::Value root;
           root["result"] = _result;
           root["flag"] = _flag;
           root["equation"] = _equation;

           Json::FastWriter writer;
           std::string str = writer.write(root);

           *out = str;
#endif
       }

       // 反序列化 -- 解析
       bool DeSerialize(const std::string &in)
       {
#ifdef SELF
           // "result flag equation"

           auto left_blank_pos = in.find(CAL_SEP);
           if (left_blank_pos == std::string::npos)
               return false;
           std::string res_str = in.substr(0, left_blank_pos);
           if (res_str.empty())
               return false;

           auto second_blank_pos = in.find(CAL_SEP, left_blank_pos + 1);
           if (second_blank_pos == std::string::npos)
               return false;
           std::string equation = in.substr(second_blank_pos + 1);
           if (equation.empty())
               return false;

           if (left_blank_pos + 1 + CAL_SEP.size() != second_blank_pos)
               return false;
           _result = std::stoi(res_str);
           _flag = in[second_blank_pos - 1] - '0';
           _equation = equation;
           return true;
#else
           Json::Value root;
           Json::Reader reader;
           if (!reader.parse(in, root))
               return false;
           _result = root["result"].asInt();
           _flag = root["flag"].asInt();
           _equation = root["equation"].asString();
           return true;
#endif
       }
       ~Response() {}

   public:
       int _result = 0;
       int _flag = 0;                         // 0表示操作符正确，1表示除0错误，2表示取模0错误，3表示操作符错误
       string _equation = "操作符不符合要求"; // 等式
   };

   const std::string opers = "+-*/%&^";

   class CalFactory
   {
   public:
       CalFactory()
       {
           srand(time(nullptr) ^ getpid() ^ 2);
       }
       void Product(Request &req)
       {
           req._x = rand() & 5 + 1;
           usleep(req._x * 20);
           req._y = rand() % 10 + 5;
           // req._y = 0; // 测试
           usleep(req._x * req._y + 20);
           req._oper = opers[(rand() % opers.size())];
       }
       ~CalFactory() {}

   private:
   };
}

可以看到这个文件我们增加了#ifdef SELF #else #endif预处理指令。

作用如下：

#ifdef SELF  
// 如果定义了宏 SELF，则编译这部分代码  
#else  
// 如果没有定义宏 SELF，则编译这部分代码  
#endif

当然了，在文件开头我们使用了#define SELF 1，也可以不使用，直接在makefile文件使用LDFLAG=-DSELF=1 # 触发SELF=1。

.PHONY:all
all:tcp_client tcp_server

LDFLAG=-DSELF=1 # 触发SELF=1

tcp_client:TcpClient.cc
	g++ -o $@ $^ $(LDFLAG) -std=c++14 -lpthread -ljsoncpp
tcp_server:Main.cc
	g++ -o $@ $^ $(LDFLAG) -std=c++14 -lpthread -ljsoncpp

.PHONY:clean
clean:
	rm -f tcp_server tcp_client

运行结果：