【C++】从零实现Json-Rpc框架（2）

JsonCpp库

1.1- Json数据格式

1.2 - JsonCpp介绍

• 序列化接口

• 反序列化接口

1.3 - Json序列化实践

JsonCpp使用

Muduo库

2.1 - Muduo库是什么

2.2 - Muduo库常见接口介绍

TcpServer类基础介绍

EventLoop类基础介绍

TcpConnection类基础介绍

TcpClient类基础介绍

Buffer类基础介绍

2.3 - Muduo库快速上手

英译汉TCP服务器

编辑

英译汉客户端

Makefile

项目汇总：uyeonashi的博客-CSDN博客

项目源码：https://gitee.com/uyeonashi/project

本篇文章是对第三方库的介绍和使用（JsonCpp库，Muduo库）！

JsonCpp库

1.1- Json数据格式

Json 是一种数据交换格式，它采用完全独立于编程语言的文本格式来存储和表示数据。

例如: 我们想表示一个同学的学生信息

C 代码表示：

char *name = "xx";
int age = 18;
float score[3] = {88.5, 99, 58};

Json 表示：

{"姓名" : "xx","年龄" : 18,"成绩" : [88.5, 99, 58],"爱好" ：{"书籍" : "西游记","运动" : "打篮球"}
}

Json 的数据类型包括对象，数组，字符串，数字等。

对象：使用花括号{ } 括起来的表示一个对象
数组：使用中括号[ ] 括起来的表示一个数组
字符串：使用常规双引号" " 括起来的表示一个字符串
数字：包括整形和浮点型，直接使用

1.2 - JsonCpp介绍

Jsoncpp 库主要是用于实现Json 格式数据的序列化和反序列化，它实现了将多个数据对象组织成为json 格式字符串，以及将Json 格式字符串解析得到多个数据对象的功能。

Jsoncpp的介绍：3个类

Json::Value类：中间数据存储类

如果要将数据进行序列化，就需要先存储到Json::Value对象当中

如果要将数据进行反序列化，就是解析后，将数据对象放入到Json::Value对象当中

先看一下Json 数据对象类的表示

class Json::Value
{Value &operator=(const Value &other); //Value重载了[]和=，因此所有的赋值和获取数据都可以通过Value& operator[](const std::string& key);//简单的方式完成 val["name"] ="xx";Value& operator[](const char* key);Value removeMember(const char* key);//移除元素const Value& operator[](ArrayIndex index) const; //val["score"][0]Value& append(const Value& value);//添加数组元素val["score"].append(88);ArrayIndex size() const;//获取数组元素个数 val["score"].size();std::string asString() const;//转string string name = val["name"].asString();const char* asCString() const;//转char* char *name = val["name"].asCString();Int asInt() const;//转int int age = val["age"].asInt();float asFloat() const;//转float float weight = val["weight"].asFloat();bool asBool() const;//转 bool bool ok = val["ok"].asBool();
};

Jsoncpp 库主要借助三个类以及其对应的少量成员函数完成序列化及反序列化

• 序列化接口

Json::StreamWriter类：用于进行数控序列化

Json::StreamWriter::write() 序列化函数

Json::StreamWriterBuilder类：Json::StreamWriter工厂类 — 用于生成Json::StreamWriter对象

class JSON_API StreamWriter 
{virtual int write(Value const& root, std::ostream* sout) = 0;
}
class JSON_API StreamWriterBuilder : public StreamWriter::Factory 
{virtual StreamWriter* newStreamWriter() const;
}

• 反序列化接口

Json::CharReader类：反序列化类

Json::CharReader::parse() 反序列化函数

Json::CharReaderBuilder类：Json::CharReader工厂类 — 用于生产Json::CharReader对象

class JSON_API CharReader 
{virtual bool parse(char const* beginDoc, char const* endDoc,Value* root, std::string* errs) = 0;
}
class JSON_API CharReaderBuilder : public CharReader::Factory 
{virtual CharReader* newCharReader() const;
}

1.3 - Json序列化实践

JsonCpp使用

#include<iostream>
#include<memory>
#include<string>
#include<sstream>
#include<jsoncpp/json/json.h>//实现数据的序列化
bool serialize(const Json::Value &val,std::string &body)
{std::stringstream ss;//先实例化一个工厂类对象Json::StreamWriterBuilder swb;//通过工厂生产派生类对象std::unique_ptr<Json::StreamWriter> sw(swb.newStreamWriter());int ret = sw->write(val,&ss);if(ret != 0){std::cout << "json serialize failed\n";return false;}body = ss.str();return true;
}//实现json的反序列化
bool unserialize(const std::string &body,Json::Value &val)
{//实例化工厂对象Json::CharReaderBuilder crb;//生产Charreader对象std::string errs;std::unique_ptr<Json::CharReader> cr(crb.newCharReader());bool ret = cr->parse(body.c_str(),body.c_str()+body.size(),&val,&errs);if(ret == false){std::cout << "Json unserialize failed: " << errs << std::endl;return false;}return true;
}int main()
{const char *name = "小明";int age = 18;const char *sex = "男";float score[3] = {88,77.5,66};Json::Value student;student["姓名"] = name;student["年龄"] = age;student["性别"] = sex;student["成绩"].append(score[0]);student["成绩"].append(score[1]);student["成绩"].append(score[2]);Json::Value fav;fav["书籍"] = "西游记";fav["运动"] = "打篮球";student["爱好"] = fav;std::string body;serialize(student,body);std::cout << body << std::endl;std::string str = R"({"姓名":"小黑","年龄":19,"成绩":[32,45.5,56]})";Json::Value stu;bool ret = unserialize(str,stu);if(ret == false)return -1;std::cout << "姓名: " << stu["姓名"].asString() << std::endl;std::cout << "年龄:"  << stu["年龄"].asInt() << std::endl;int sz = stu["成绩"].size();for(int i = 0; i < sz; i++){std::cout << "成绩:" << stu["成绩"][i].asFloat() << std::endl;} return 0;
}

编译运行程序查看序列化和反序列化结果

并在在这将其封装了一下，方便我们后边的使用

Muduo库

2.1 - Muduo库是什么

Muduo由陈硕大佬开发，是一个基于非阻塞IO和事件驱动的C++高并发TCP网络编程库。 它是一款基于主从Reactor模型的网络库，其使用的线程模型是one loop per thread, 所谓one loop per thread指的是：

一个线程只能有一个事件循环（EventLoop），用于响应计时器和IO事件
一个文件描述符只能由一个线程进行读写，换句话说就是一个TCP连接必须归属于某EventLoop管理

2.2 - Muduo库常见接口介绍

TcpServer类基础介绍

TcpServer{

void start(); //启动服务

setConnectionCallback(); //设置连接建立 / 关闭时的回调函数

setMessageCallback();//设置消息处理回调函数

};

typedef std::shared_ptr<TcpConnection> TcpConnectionPtr;
typedef std::function<void (const TcpConnectionPtr&)> ConnectionCallback;
typedef std::function<void (const TcpConnectionPtr&,Buffer*,Timestamp)> MessageCallback;
class InetAddress : public muduo::copyable
{
public:InetAddress(StringArg ip, uint16_t port, bool ipv6 = false);
};class TcpServer : noncopyable
{
public:enum Option{kNoReusePort,kReusePort,};TcpServer(EventLoop* loop,const InetAddress& listenAddr,const string& nameArg,Option option = kNoReusePort);void setThreadNum(int numThreads);void start();/// 当一个新连接建立成功的时候被调用void setConnectionCallback(const ConnectionCallback& cb){ connectionCallback_ = cb; }/// 消息的业务处理回调函数---这是收到新连接消息的时候被调用的函数void setMessageCallback(const MessageCallback& cb){ messageCallback_ = cb; }
};

EventLoop类基础介绍

EvenLoop{

void loop(); //开始事件监控事件循环

void quit(); //停止循环

Timerld runAfter(delay,cb); //定时任务

};

class EventLoop : noncopyable
{
public:// Loops forever.// Must be called in the same thread as creation of the object.void loop();// Quits loop.// This is not 100% thread safe, if you call through a raw pointer,///better to call through shared_ptr<EventLoop> for 100% safety.void quit();TimerId runAt(Timestamp time, TimerCallback cb);// Runs callback after @c delay seconds.// Safe to call from other threads.TimerId runAfter(double delay, TimerCallback cb);// Runs callback every @c interval seconds.// Safe to call from other threads.TimerId runEvery(double interval, TimerCallback cb);// Cancels the timer.// Safe to call from other threads.void cancel(TimerId timerId);
private:std::atomic<bool> quit_;std::unique_ptr<Poller> poller_;mutable MutexLock mutex_;std::vector<Functor> pendingFunctors_ GUARDED_BY(mutex_);
};

TcpConnection类基础介绍

TcpConnection{

void send(std::string &msg); //发送数据

bool connected(); //判断当前连接是否连接正常

shutdown down(); //关闭连接

}

class TcpConnection : noncopyable,public std::enable_shared_from_this<TcpConnection>
{
public:
// Constructs a TcpConnection with a connected sockfd
//
// User should not create this object.TcpConnection(EventLoop* loop,const string& name,int sockfd,const InetAddress& localAddr,const InetAddress& peerAddr);bool connected() const { return state_ == kConnected; }bool disconnected() const { return state_ == kDisconnected; }void send(string&& message); // C++11void send(const void* message, int len);void send(const StringPiece& message);// void send(Buffer&& message); // C++11void send(Buffer* message); // this one will swap datavoid shutdown(); // NOT thread safe, no simultaneous callingvoid setContext(const boost::any& context){ context_ = context; }const boost::any& getContext() const{ return context_; }boost::any* getMutableContext(){ return &context_; }void setConnectionCallback(const ConnectionCallback& cb){ connectionCallback_ = cb; }void setMessageCallback(const MessageCallback& cb){ messageCallback_ = cb; }
private:enum StateE { kDisconnected, kConnecting, kConnected, kDisconnecting };EventLoop* loop_;ConnectionCallback connectionCallback_;MessageCallback messageCallback_;WriteCompleteCallback writeCompleteCallback_;boost::any context_;
};

TcpClient类基础介绍

TcpClient{

void connect(); //连接服务器

void disconnect(); //关闭连接

TcpConnectionPtr connection() const; // 获取客户端对应的TcpConnection连接

Muduo库的客户端也是通过Eventloop进行IO事件监控IO处理的

void setConnectionCallback(ConnectionCallback cb); //连接建立成功 / 关闭的回调处理

void setMessageCallback(MessageCallback cb); //收到消息的回调处理

}

CountDownLatch{ //做计数同步操作的类

void wait(); //计数大于0则阻塞

countDown(); //计数--，为0时唤醒wait

}

因为Client的connect接口是一个非阻塞操作，所以可能出现移走以外情况：connect连接还没有完全建立的情况下，调用connection接口获取连接，send发送数据

class TcpClient : noncopyable
{
public:// TcpClient(EventLoop* loop);// TcpClient(EventLoop* loop, const string& host, uint16_t port);TcpClient(EventLoop* loop,const InetAddress& serverAddr,const string& nameArg);~TcpClient(); // force out-line dtor, for std::unique_ptr members.void connect();//连接服务器void disconnect();//关闭连接void stop();//获取客⼾端对应的通信连接Connection对象的接⼝，发起connect后，有可能还没有连接建⽴成功TcpConnectionPtr connection() const{MutexLockGuard lock(mutex_);return connection_;}// 连接服务器成功时的回调函数void setConnectionCallback(ConnectionCallback cb){ connectionCallback_ = std::move(cb); }// 收到服务器发送的消息时的回调函数void setMessageCallback(MessageCallback cb){ messageCallback_ = std::move(cb); }
private:EventLoop* loop_;ConnectionCallback connectionCallback_;MessageCallback messageCallback_;WriteCompleteCallback writeCompleteCallback_;TcpConnectionPtr connection_ GUARDED_BY(mutex_);
};/*需要注意的是，因为muduo库不管是服务端还是客⼾端都是异步操作，对于客⼾端来说如果我们在连接还没有完全建⽴成功的时候发送数据，这是不被允许的。因此我们可以使⽤内置的CountDownLatch类进⾏同步控制*/class CountDownLatch : noncopyable{public:explicit CountDownLatch(int count);void wait(){MutexLockGuard lock(mutex_);while (count_ > 0){condition_.wait();}}void countDown(){MutexLockGuard lock(mutex_);--count_;if (count_ == 0){condition_.notifyAll();}}int getCount() const;private:mutable MutexLock mutex_;Condition condition_ GUARDED_BY(mutex_);int count_ GUARDED_BY(mutex_);};

Buffer类基础介绍

Buffer {

size_t readableBytes(); //获取缓冲区可读数据大小

const char* peek(); // 获取缓冲区的起始地址

int32_t peeklnt32() const； //尝试从缓冲区获取4字节数据，进行网络字节序转换为整形，但数据并不从缓冲区删除

void retrieventlnt32(); //数据读取位置向后偏移4字节,本质上就是删除起始位置的4字节数据

int32_t readlnt32();

string retrieveAllAsString(); //从缓冲区取出所有数据，当做string返回，并删除缓冲区中的数据

string retrieveAsString(size_t len),从缓冲区中取出len长度的数据，当做string返回，并删除缓冲区中的数据

}

class Buffer : public muduo::copyable
{public:static const size_t kCheapPrepend = 8;static const size_t kInitialSize = 1024;explicit Buffer(size_t initialSize = kInitialSize): buffer_(kCheapPrepend + initialSize),readerIndex_(kCheapPrepend),writerIndex_(kCheapPrepend);void swap(Buffer& rhs)size_t readableBytes() constsize_t writableBytes() constconst char* peek() constconst char* findEOL() constconst char* findEOL(const char* start) constvoid retrieve(size_t len)void retrieveInt64()void retrieveInt32()void retrieveInt16()void retrieveInt8()string retrieveAllAsString()string retrieveAsString(size_t len)void append(const StringPiece& str)void append(const char* /*restrict*/ data, size_t len)void append(const void* /*restrict*/ data, size_t len)char* beginWrite()const char* beginWrite() constvoid hasWritten(size_t len)void appendInt64(int64_t x)void appendInt32(int32_t x)void appendInt16(int16_t x)void appendInt8(int8_t x)int64_t readInt64()int32_t readInt32()int16_t readInt16()int8_t readInt8()int64_t peekInt64() constint16_t peekInt16() constint8_t peekInt8() constvoid prependInt64(int64_t x)void prependInt32(int32_t x)void prependInt16(int16_t x)void prependInt8(int8_t x)void prepend(const void* /*restrict*/ data, size_t len)
private:std::vector<char> buffer_;size_t readerIndex_;size_t writerIndex_;static const char kCRLF[];
};

2.3 - Muduo库快速上手

我们使用Muduo网络库来实现一个简单英译汉服务器和客户端快速上手Muduo库

英译汉TCP服务器

包含头文件时指定路径

// 实现一个翻译服务器，客户端发来一个英语单词，返回一个汉语词典#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <iostream>
#include <string>
#include <unordered_map>class DictServer
{
public:DictServer(int port): _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "DictServer", muduo::net::TcpServer::kReusePort){//设置连接事件（连接建立/管理）的回调_server.setConnectionCallback(std::bind(&DictServer::onConnection, this, std::placeholders::_1));//设置连接消息的回调_server.setMessageCallback(std::bind(&DictServer::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));}void start(){_server.start(); //先开始监听_baseloop.loop();//开始死循环执行事件监控}private:void onConnection(const muduo::net::TcpConnectionPtr &conn){if (conn->connected())std::cout << "连接建立!\n";elsestd::cout << "连接断开!\n";}void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp){static std::unordered_map<std::string, std::string> dict_map = {{"hello", "你好"},{"world", "世界"},{"bite", "比特"}};std::string msg = buf->retrieveAllAsString();std::string res;auto it = dict_map.find(msg);if (it != dict_map.end()){res = it->second;}else{res = "未知单词";}conn->send(res);}private:muduo::net::EventLoop _baseloop;muduo::net::TcpServer _server;
};int main()
{DictServer server(9090);server.start();return 0;
}

英译汉客户端

#include <muduo/net/TcpClient.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/net/Buffer.h>
#include <muduo/base/CountDownLatch.h>
#include <memory>
#include <iostream>
#include <string>class DictClient
{
public:DictClient(const std::string &sip, int sport):_baseloop(_loopthread.startLoop()) ,_downlatch(1), _client(_baseloop, muduo::net::InetAddress(sip, sport), "DictClient"){// 设置连接事件（连接建立/管理）的回调_client.setConnectionCallback(std::bind(&DictClient::onConnection, this, std::placeholders::_1));// 设置连接消息的回调_client.setMessageCallback(std::bind(&DictClient::onMessage, this,std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));//连接服务器_client.connect();_downlatch.wait();}bool send(const std::string &msg){if(_conn->connected() == false){std::cout << "链接已断开，发送数据失败！\n";return false;}_conn->send(msg);//_baseloop.loop();  //开始事件循环监控 -- 内部是个死循环，客户端不能直接使用return true;}private:void onConnection(const muduo::net::TcpConnectionPtr &conn){if (conn->connected()){std::cout << "连接建立!\n";_downlatch.countDown(); // 计数--，为0时唤醒阻塞_conn = conn;}else{std::cout << "连接断开!\n";_conn.reset();}}void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp){std::string res = buf->retrieveAllAsString();std::cout << res << std::endl;}private:muduo::net::TcpConnectionPtr _conn;muduo::CountDownLatch _downlatch;muduo::net::EventLoopThread _loopthread;muduo::net::EventLoop *_baseloop;muduo::net::TcpClient _client;
};int main()
{DictClient client("127.0.0.1",9090);while(1){std::string msg;std::cin >> msg;client.send(msg);}return 0;
}

Makefile

CFLAG=-I ../../build/release-install-cpp11/include/
LFLAF=-L ../../build/release-install-cpp11/lib -lmuduo_net -lmuduo_base -pthread
all: server client
server:server.cppg++ $(CFLAG) -o $@ $^ $(LFLAF) -std=c++11
client:client.cppg++ $(CFLAG) -o $@ $^ $(LFLAF) -std=c++11

C++11 异步操作

3.1 - std::future

介绍

std::future是C++11标准库中的一个模板类，它表示一个异步操作的结果。当我们在多线程编程中使用异步任务时，std::future可以帮助我们在需要的时候获取任务的执行结果。std::future的一个重要特性是能够阻塞当前线程，直到异步操作完成，从而确保我们在获取结果时不会遇到未完成的操作。

应用场景

异步任务： 当我们需要在后台执行一些耗时操作时，如网络请求或计算密集型任务等，std::future可以用来表示这些异步任务的结果。通过将任务与主线程分离，我们可以实现任务的并行处理，从而提高程序的执行效率
并发控制： 在多线程编程中，我们可能需要等待某些任务完成后才能继续执行其他操作。通过使用std::future，我们可以实现线程之间的同步，确保任务完成后再获取结果并继续执行后续操作
结果获取：std::future提供了一种安全的方式来获取异步任务的结果。我们可以使用std::future::get()函数来获取任务的结果，此函数会阻塞当前线程，直到异步操作完成。这样，在调用get()函数时，我们可以确保已经获取到了所需的结果

用法示例

使用 std::async关联异步任务

std::async是一种将任务与std::future关联的简单方法。它创建并运行一个异步任务，并返回一个与该任务结果关联的std::future对象。默认情况下，std::async是否启动一个新线程，或者在等待future时，任务是否同步运行都取决于你给的参数。这个参数为std::launch类型：

  std::launch::deferred 表明该函数会被延迟调用，直到在future上调用get()或者wait()才会开始执行任务
        std::launch::async 表明函数会在自己创建的线程上运行
        std::launch::deferred | std::launch::async 内部通过系统等条件自动选择策略

#include <iostream>
#include <thread>
#include <future>int Add(int num1, int num2) 
{std::cout << "into add!\n";return num1 + num2;
}int main()
{//std::launch::async策略：内部创建一个线程执行函数，函数运行结果通过future获取//std::launch::deferred策略：同步策略，获取结果的时候再去执行任务//std::future<int> res = std::async(std::launch::deferred, Add, 11, 22);std::future<int> res = std::async(std::launch::async, Add, 11, 22);//进行了一个异步非阻塞调用std::this_thread::sleep_for(std::chrono::seconds(1));std::cout << "--------------------------\n";//std::future<int>::get()  用于获取异步任务的结果，如果还没有结果就会阻塞std::cout << res.get() << std::endl;return 0;
}

使用std::packaged_task和std::future配合

std::packaged_task就是将任务和 std::feature 绑定在一起的模板，是一种对任务的封装。我们可以通过std::packaged_task对象获取任务相关联的std::feature对象，通过调用get_future()方法获得。
std::packaged_task的模板参数是函数签名。
可以把std::future和std::async看成是分开的，而 std::packaged_task则是一个整体。

#include<iostream>
#include<future>
#include<thread>
#include<memory>int Add(int num1, int num2)
{std::cout << "into add!\n";return num1 + num2;
}int main()
{//1. 封装任务auto task = std::make_shared<std::packaged_task<int(int, int)>>(Add);//2. 获取任务包关联的future对象std::future<int> res = task->get_future();std::thread thr([&task](){(*task)(11,22);});//4. 获取结果std::cout << res.get() << std::endl;thr.join();return 0;
}

使用std::promise和std::future配合

std::promise提供了一种设置值的方式，它可以在设置之后通过相关联的std::future对象进行读取。换种说法就是之前说过std::future可以读取一个异步函数的返回值了，但是要等待就绪，而std::promise就提供一种方式手动让 std::future就绪

#include<iostream>
#include<future>
#include<thread>
#include<memory>int Add(int num1,int num2)
{std::cout << "into add!\n";return num1+num2;
}int main()
{//1.在使用的时候，先实例化一个指定结果的promise对象std::promise<int> pro;//2. 通过promise对象获取关联future对象std::future<int> res = pro.get_future();//3. 在任意位置给promise设置数据，就可以通过关联的future获取到这个设置的数据了std::thread thr([&pro](){int sum = Add(11,22);pro.set_value(sum);});std::cout << res.get() << std::endl;thr.join();return 0;
}